Bulk API: Improve memory usage when executing large bulk requests, closes elastic#724.

Author: kimchy

modules/elasticsearch/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

@@ -35,11 +35,9 @@
 import org.elasticsearch.cluster.metadata.MappingMetaData;
 import org.elasticsearch.cluster.routing.ShardIterator;
 import org.elasticsearch.common.Strings;
-import org.elasticsearch.common.collect.Sets;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.engine.Engine;
-import org.elasticsearch.index.engine.EngineException;
 import org.elasticsearch.index.mapper.DocumentMapper;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.ParsedDocument;
@@ -54,7 +52,6 @@
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
-import java.util.Set;
 
 /**
  * Performs the index operation.
@@ -106,11 +103,14 @@ public class TransportShardBulkAction extends TransportShardReplicationOperation
     }
 
     @Override protected PrimaryResponse<BulkShardResponse> shardOperationOnPrimary(ClusterState clusterState, ShardOperationRequest shardRequest) {
-        IndexShard indexShard = indexShard(shardRequest);
         final BulkShardRequest request = shardRequest.request;
-        BulkItemResponse[] responses = new BulkItemResponse[request.items().length];
+        IndexShard indexShard = indexShard(shardRequest);
+
         Engine.Operation[] ops = new Engine.Operation[request.items().length];
-        for (int i = 0; i < ops.length; i++) {
+
+
+        BulkItemResponse[] responses = new BulkItemResponse[request.items().length];
+        for (int i = 0; i < request.items().length; i++) {
             BulkItemRequest item = request.items()[i];
             if (item.request() instanceof IndexRequest) {
                 IndexRequest indexRequest = (IndexRequest) item.request();
@@ -126,11 +126,39 @@ public class TransportShardBulkAction extends TransportShardReplicationOperation
 
                     SourceToParse sourceToParse = SourceToParse.source(indexRequest.source()).type(indexRequest.type()).id(indexRequest.id())
                             .routing(indexRequest.routing()).parent(indexRequest.parent());
+                    long version;
+                    ParsedDocument doc;
+                    Engine.Operation op;
                     if (indexRequest.opType() == IndexRequest.OpType.INDEX) {
-                        ops[i] = indexShard.prepareIndex(sourceToParse).version(indexRequest.version()).origin(Engine.Operation.Origin.PRIMARY);
+                        Engine.Index index = indexShard.prepareIndex(sourceToParse).version(indexRequest.version()).origin(Engine.Operation.Origin.PRIMARY);
+                        doc = indexShard.index(index);
+                        version = index.version();
+                        op = index;
                     } else {
-                        ops[i] = indexShard.prepareCreate(sourceToParse).version(indexRequest.version()).origin(Engine.Operation.Origin.PRIMARY);
+                        Engine.Create create = indexShard.prepareCreate(sourceToParse).version(indexRequest.version()).origin(Engine.Operation.Origin.PRIMARY);
+                        doc = indexShard.create(create);
+                        version = create.version();
+                        op = create;
+                    }
+                    // update the version on request so it will happen on the replicas
+                    indexRequest.version(version);
+
+                    // update mapping on master if needed, we won't update changes to the same type, since once its changed, it won't have mappers added
+                    if (doc.mappersAdded()) {
+                        updateMappingOnMaster(indexRequest);
+                    }
+
+                    // if we are going to percolate, then we need to keep this op for the postPrimary operation
+                    if (Strings.hasLength(indexRequest.percolate())) {
+                        if (ops == null) {
+                            ops = new Engine.Operation[request.items().length];
+                        }
+                        ops[i] = op;
                     }
+
+                    // add the response
+                    responses[i] = new BulkItemResponse(item.id(), indexRequest.opType().toString().toLowerCase(),
+                            new IndexResponse(indexRequest.index(), indexRequest.type(), indexRequest.id(), version));
                 } catch (Exception e) {
                     if (logger.isDebugEnabled()) {
                         logger.debug("[" + shardRequest.request.index() + "][" + shardRequest.shardId + "]" + ": Failed to execute bulk item (index) [" + indexRequest + "]", e);
@@ -141,7 +169,14 @@ public class TransportShardBulkAction extends TransportShardReplicationOperation
             } else if (item.request() instanceof DeleteRequest) {
                 DeleteRequest deleteRequest = (DeleteRequest) item.request();
                 try {
-                    ops[i] = indexShard.prepareDelete(deleteRequest.type(), deleteRequest.id(), deleteRequest.version()).origin(Engine.Operation.Origin.PRIMARY);
+                    Engine.Delete delete = indexShard.prepareDelete(deleteRequest.type(), deleteRequest.id(), deleteRequest.version()).origin(Engine.Operation.Origin.PRIMARY);
+                    indexShard.delete(delete);
+                    // update the request with teh version so it will go to the replicas
+                    deleteRequest.version(delete.version());
+
+                    // add the response
+                    responses[i] = new BulkItemResponse(item.id(), "delete",
+                            new DeleteResponse(deleteRequest.index(), deleteRequest.type(), deleteRequest.id(), delete.version(), delete.notFound()));
                 } catch (Exception e) {
                     if (logger.isDebugEnabled()) {
                         logger.debug("[" + shardRequest.request.index() + "][" + shardRequest.shardId + "]" + ": Failed to execute bulk item (delete) [" + deleteRequest + "]", e);
@@ -152,61 +187,11 @@ public class TransportShardBulkAction extends TransportShardReplicationOperation
             }
         }
 
-        EngineException[] failures = indexShard.bulk(new Engine.Bulk(ops).refresh(request.refresh()));
-        // process failures and mappings
-        Set<String> processedTypes = Sets.newHashSet();
-        for (int i = 0; i < ops.length; i++) {
-            // failed to parse, already set the failure, skip
-            if (ops[i] == null) {
-                continue;
-            }
-
-            BulkItemRequest item = request.items()[i];
-            if (item.request() instanceof IndexRequest) {
-                IndexRequest indexRequest = (IndexRequest) item.request();
-                long version;
-                if (indexRequest.opType() == IndexRequest.OpType.INDEX) {
-                    Engine.Index engineIndex = (Engine.Index) ops[i];
-                    version = engineIndex.version();
-                    if (!processedTypes.contains(engineIndex.type())) {
-                        processedTypes.add(engineIndex.type());
-                        ParsedDocument doc = engineIndex.parsedDoc();
-                        if (doc.mappersAdded()) {
-                            updateMappingOnMaster(indexRequest);
-                        }
-                    }
-                } else {
-                    Engine.Create engineCreate = (Engine.Create) ops[i];
-                    version = engineCreate.version();
-                    if (!processedTypes.contains(engineCreate.type())) {
-                        processedTypes.add(engineCreate.type());
-                        ParsedDocument doc = engineCreate.parsedDoc();
-                        if (doc.mappersAdded()) {
-                            updateMappingOnMaster(indexRequest);
-                        }
-                    }
-                }
-                // update the version on request so it will happen on the replicas
-                indexRequest.version(version);
-                if (failures != null && failures[i] != null) {
-                    responses[i] = new BulkItemResponse(item.id(), indexRequest.opType().toString().toLowerCase(),
-                            new BulkItemResponse.Failure(indexRequest.index(), indexRequest.type(), indexRequest.id(), ExceptionsHelper.detailedMessage(failures[i])));
-                } else {
-                    responses[i] = new BulkItemResponse(item.id(), indexRequest.opType().toString().toLowerCase(),
-                            new IndexResponse(indexRequest.index(), indexRequest.type(), indexRequest.id(), version));
-                }
-            } else if (item.request() instanceof DeleteRequest) {
-                DeleteRequest deleteRequest = (DeleteRequest) item.request();
-                Engine.Delete engineDelete = (Engine.Delete) ops[i];
-                // update the version on request so it will happen on the replicas
-                deleteRequest.version(engineDelete.version());
-                if (failures != null && failures[i] != null) {
-                    responses[i] = new BulkItemResponse(item.id(), "delete",
-                            new BulkItemResponse.Failure(deleteRequest.index(), deleteRequest.type(), deleteRequest.id(), ExceptionsHelper.detailedMessage(failures[i])));
-                } else {
-                    responses[i] = new BulkItemResponse(item.id(), "delete",
-                            new DeleteResponse(deleteRequest.index(), deleteRequest.type(), deleteRequest.id(), engineDelete.version(), engineDelete.notFound()));
-                }
+        if (request.refresh()) {
+            try {
+                indexShard.refresh(new Engine.Refresh(false));
+            } catch (Exception e) {
+                // ignore
             }
         }
         BulkShardResponse response = new BulkShardResponse(new ShardId(request.index(), request.shardId()), responses);
@@ -223,6 +208,9 @@ public class TransportShardBulkAction extends TransportShardReplicationOperation
                 // failure, continue
                 continue;
             }
+            if (ops[i] == null) {
+                continue; // failed
+            }
             if (itemRequest.request() instanceof IndexRequest) {
                 IndexRequest indexRequest = (IndexRequest) itemRequest.request();
                 if (!Strings.hasLength(indexRequest.percolate())) {
@@ -247,33 +235,41 @@ public class TransportShardBulkAction extends TransportShardReplicationOperation
     @Override protected void shardOperationOnReplica(ShardOperationRequest shardRequest) {
         IndexShard indexShard = indexShard(shardRequest);
         final BulkShardRequest request = shardRequest.request;
-        Engine.Operation[] ops = new Engine.Operation[request.items().length];
-        for (int i = 0; i < ops.length; i++) {
+        for (int i = 0; i < request.items().length; i++) {
             BulkItemRequest item = request.items()[i];
             if (item.request() instanceof IndexRequest) {
                 IndexRequest indexRequest = (IndexRequest) item.request();
                 try {
                     SourceToParse sourceToParse = SourceToParse.source(indexRequest.source()).type(indexRequest.type()).id(indexRequest.id())
                             .routing(indexRequest.routing()).parent(indexRequest.parent());
                     if (indexRequest.opType() == IndexRequest.OpType.INDEX) {
-                        ops[i] = indexShard.prepareIndex(sourceToParse).version(indexRequest.version()).origin(Engine.Operation.Origin.REPLICA);
+                        Engine.Index index = indexShard.prepareIndex(sourceToParse).version(indexRequest.version()).origin(Engine.Operation.Origin.REPLICA);
+                        indexShard.index(index);
                     } else {
-                        ops[i] = indexShard.prepareCreate(sourceToParse).version(indexRequest.version()).origin(Engine.Operation.Origin.REPLICA);
+                        Engine.Create create = indexShard.prepareCreate(sourceToParse).version(indexRequest.version()).origin(Engine.Operation.Origin.REPLICA);
+                        indexShard.create(create);
                     }
                 } catch (Exception e) {
                     // ignore, we are on backup
                 }
             } else if (item.request() instanceof DeleteRequest) {
                 DeleteRequest deleteRequest = (DeleteRequest) item.request();
                 try {
-                    ops[i] = indexShard.prepareDelete(deleteRequest.type(), deleteRequest.id(), deleteRequest.version()).origin(Engine.Operation.Origin.REPLICA);
+                    Engine.Delete delete = indexShard.prepareDelete(deleteRequest.type(), deleteRequest.id(), deleteRequest.version()).origin(Engine.Operation.Origin.REPLICA);
+                    indexShard.delete(delete);
                 } catch (Exception e) {
                     // ignore, we are on backup
                 }
             }
         }
 
-        indexShard.bulk(new Engine.Bulk(ops).refresh(request.refresh()));
+        if (request.refresh()) {
+            try {
+                indexShard.refresh(new Engine.Refresh(false));
+            } catch (Exception e) {
+                // ignore
+            }
+        }
     }
 
     private void updateMappingOnMaster(final IndexRequest request) {

modules/elasticsearch/src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -54,8 +54,6 @@ public interface Engine extends IndexShardComponent, CloseableComponent {
      */
     void start() throws EngineException;
 
-    EngineException[] bulk(Bulk bulk) throws EngineException;
-
     void create(Create create) throws EngineException;
 
     void index(Index index) throws EngineException;
@@ -264,28 +262,6 @@ static enum Origin {
         Origin origin();
     }
 
-    static class Bulk {
-        private final Operation[] ops;
-        private boolean refresh;
-
-        public Bulk(Operation[] ops) {
-            this.ops = ops;
-        }
-
-        public Operation[] ops() {
-            return this.ops;
-        }
-
-        public boolean refresh() {
-            return refresh;
-        }
-
-        public Bulk refresh(boolean refresh) {
-            this.refresh = refresh;
-            return this;
-        }
-    }
-
     static class Create implements Operation {
         private final Term uid;
         private final ParsedDocument doc;

modules/elasticsearch/src/main/java/org/elasticsearch/index/engine/robin/RobinEngine.java

@@ -219,77 +219,6 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine,
         return refreshInterval;
     }
 
-    @Override public EngineException[] bulk(Bulk bulk) throws EngineException {
-        EngineException[] failures = null;
-        rwl.readLock().lock();
-        try {
-            IndexWriter writer = this.indexWriter;
-            if (writer == null) {
-                throw new EngineClosedException(shardId);
-            }
-            for (int i = 0; i < bulk.ops().length; i++) {
-                Operation op = bulk.ops()[i];
-                if (op == null) {
-                    continue;
-                }
-                try {
-                    switch (op.opType()) {
-                        case CREATE:
-                            Create create = (Create) op;
-                            innerCreate(create, writer);
-                            break;
-                        case INDEX:
-                            Index index = (Index) op;
-                            innerIndex(index, writer);
-                            break;
-                        case DELETE:
-                            Delete delete = (Delete) op;
-                            innerDelete(delete, writer);
-                            break;
-                    }
-                } catch (Exception e) {
-                    if (failures == null) {
-                        failures = new EngineException[bulk.ops().length];
-                    }
-                    switch (op.opType()) {
-                        case CREATE:
-                            if (e instanceof EngineException) {
-                                failures[i] = (EngineException) e;
-                            } else {
-                                failures[i] = new CreateFailedEngineException(shardId, (Create) op, e);
-                            }
-                            break;
-                        case INDEX:
-                            if (e instanceof EngineException) {
-                                failures[i] = (EngineException) e;
-                            } else {
-                                failures[i] = new IndexFailedEngineException(shardId, (Index) op, e);
-                            }
-                            break;
-                        case DELETE:
-                            if (e instanceof EngineException) {
-                                failures[i] = (EngineException) e;
-                            } else {
-                                failures[i] = new DeleteFailedEngineException(shardId, (Delete) op, e);
-                            }
-                            break;
-                    }
-                }
-            }
-            dirty = true;
-            if (bulk.refresh()) {
-                try {
-                    refresh(new Refresh(false));
-                } catch (Exception e) {
-                    //ignore
-                }
-            }
-        } finally {
-            rwl.readLock().unlock();
-        }
-        return failures;
-    }
-
     @Override public void create(Create create) throws EngineException {
         rwl.readLock().lock();
         try {

modules/elasticsearch/src/main/java/org/elasticsearch/index/shard/service/IndexShard.java

@@ -62,8 +62,6 @@ public interface IndexShard extends IndexShardComponent {
 
     void delete(Engine.Delete delete) throws ElasticSearchException;
 
-    EngineException[] bulk(Engine.Bulk bulk) throws ElasticSearchException;
-
     void deleteByQuery(byte[] querySource, @Nullable String queryParserName, String... types) throws ElasticSearchException;
 
     byte[] get(String type, String id) throws ElasticSearchException;

modules/elasticsearch/src/main/java/org/elasticsearch/index/shard/service/InternalIndexShard.java

@@ -303,42 +303,6 @@ public InternalIndexShard start(String reason) throws IndexShardStartedException
         engine.delete(delete);
     }
 
-    @Override public EngineException[] bulk(Engine.Bulk bulk) throws ElasticSearchException {
-        writeAllowed();
-        if (listeners != null) {
-            for (int i = 0; i < bulk.ops().length; i++) {
-                Engine.Operation op = bulk.ops()[i];
-                if (op == null) {
-                    continue;
-                }
-                switch (op.opType()) {
-                    case CREATE:
-                        Engine.Create create = (Engine.Create) op;
-                        for (OperationListener listener : listeners) {
-                            bulk.ops()[i] = listener.beforeCreate(create);
-                        }
-                        break;
-                    case INDEX:
-                        Engine.Index index = (Engine.Index) op;
-                        for (OperationListener listener : listeners) {
-                            bulk.ops()[i] = listener.beforeIndex(index);
-                        }
-                        break;
-                    case DELETE:
-                        Engine.Delete delete = (Engine.Delete) op;
-                        for (OperationListener listener : listeners) {
-                            bulk.ops()[i] = listener.beforeDelete(delete);
-                        }
-                        break;
-                }
-            }
-        }
-        if (logger.isTraceEnabled()) {
-            logger.trace("bulk, items [{}]", bulk.ops().length);
-        }
-        return engine.bulk(bulk);
-    }
-
     @Override public void deleteByQuery(byte[] querySource, @Nullable String queryParserName, String... types) throws ElasticSearchException {
         writeAllowed();
         if (types == null) {