PUT movies
{
  "mappings": {
    "_doc": {
      "dynamic": "false" // true, false, strict
    }
  }
}

GET /megacrp/_mapping
GET /megacrp/employee/_mapping

PUT /website/blog/123
{
  "title": "My first blog entry",
  "text": "Just trying this out...",
  "date": "2014/01/01"
}

# 指定ID的index操作，其实是个upsert操作
PUT /website/blog/123?op_type=create
{ ... }
# 创建一个文档
PUT /website/blog/123/_create
{ ... }

POST /website/blog/
{
  "title": "My second blog entry",
  "text": "Still trying this out...",
  "date": "2014/01/01"
}

# 文档必须已经存在
POST /website/blog/1/_update
{
   "doc" : {
      "tags" : [ "testing" ],
      "views": 0
   }
}

POST /website/blog/zVmOW2EBsZ0GEqF92yf6/_update
{
   "script" : {
      "source" : "if(ctx._source.views == params.count) { ctx.op = 'delete'} ctx._source.views+=1; ctx._source.tags.add(params.new_tag)",
      "params" : {
        "new_tag" : "search",
        "count": 2
      }
    },
    "upsert": {
        "views": 1
    }
}

POST /website/blog/zVmOW2EBsZ0GEqF92yf6/_update?retry_on_conflict=5 
{
   "script" : "ctx._source.views+=1",
   "upsert": {
       "views": 0
   }
}

GET /website/blog/123?pretty
GET /website/blog/123/_source
GET /website/blog/123?_source=title,text

GET /_mget
{
   "docs" : [
      {
         "_index" : "website",
         "_type" : "blog",
         "_id" : "zVmOW2EBsZ0GEqF92yf6"
      },
      {
         "_index" : "website",
         "_type" : "blog",
         "_id" : 1,
         "_source": "views"
      }
   ]
}
GET /website/blog/_mget
{
   "ids" : [ "2", "1" ]
}

HEAD /website/blog/124

DELETE /website/blog/123

{ action: { metadata }}\n
{ request body }\n
{ action: { metadata }}\n
{ request body }\n

POST /_bulk
{ "delete": { "_index": "website", "_type": "blog", "_id": "123" }} 
{ "create": { "_index": "website", "_type": "blog", "_id": "123" }}
{ "title": "My first blog post" }
{ "index": { "_index": "website", "_type": "blog" }}
{ "title": "My second blog post" }
{ "update": { "_index": "webiite", "_type": "blog", "_id": "123", "_retry_on_conflict" : 3} }
{ "doc" : {"title" : "My updated blog post"} } 

actions.register(BulkAction.INSTANCE, TransportBulkAction.class,
        TransportShardBulkAction.class);

registerHandler.accept(new RestBulkAction(settings, restController));

{
  "title": "My first blog entry",
  "text": "Just trying this out...",
  "date": "2014/01/01"
}

for (String index : indices) {
    boolean shouldAutoCreate;
    try {
        shouldAutoCreate = shouldAutoCreate(index, state);
    } catch (IndexNotFoundException e) {
        shouldAutoCreate = false;
        indicesThatCannotBeCreated.put(index, e);
    }
    if (shouldAutoCreate) {
        autoCreateIndices.add(index);
    }
}

void createIndex(String index, TimeValue timeout, ActionListener<CreateIndexResponse> listener) {
    CreateIndexRequest createIndexRequest = new CreateIndexRequest();
    createIndexRequest.index(index);
    createIndexRequest.cause("auto(bulk api)");
    createIndexRequest.masterNodeTimeout(timeout);
    createIndexAction.execute(createIndexRequest, listener);
}

switch (docWriteRequest.opType()) {
    // 创建索引、mapping请求
    case CREATE:
    case INDEX:
        IndexRequest indexRequest = (IndexRequest) docWriteRequest;
        final IndexMetaData indexMetaData = metaData.index(concreteIndex);
        MappingMetaData mappingMd = indexMetaData.mappingOrDefault(indexRequest.type());
        Version indexCreated = indexMetaData.getCreationVersion();
        indexRequest.resolveRouting(metaData);
        indexRequest.process(indexCreated, mappingMd, concreteIndex.getName());
        break;
    // 更新文档请求
    case UPDATE:
        TransportUpdateAction.resolveAndValidateRouting(metaData, concreteIndex.getName(), (UpdateRequest) docWriteRequest);
        break;
    // 删除文档操作
    case DELETE:
        docWriteRequest.routing(metaData.resolveWriteIndexRouting(docWriteRequest.parent(), docWriteRequest.routing(), docWriteRequest.index()));
        // check if routing is required, if so, throw error if routing wasn't specified
        if (docWriteRequest.routing() == null && metaData.routingRequired(concreteIndex.getName(), docWriteRequest.type())) {
            throw new RoutingMissingException(concreteIndex.getName(), docWriteRequest.type(), docWriteRequest.id());
        }
        break;
    default: throw new AssertionError("request type not supported: [" + docWriteRequest.opType() + "]");
}

// 根据文档ID分配给对应的shardId
// first, go over all the requests and create a ShardId -> Operations mapping
Map<ShardId, List<BulkItemRequest>> requestsByShard = new HashMap<>();
for (int i = 0; i < bulkRequest.requests.size(); i++) {
    DocWriteRequest request = bulkRequest.requests.get(i);
    if (request == null) {
        continue;
    }
    String concreteIndex = concreteIndices.getConcreteIndex(request.index()).getName();
    ShardId shardId = clusterService.operationRouting().indexShards(clusterState, concreteIndex, request.id(), request.routing()).shardId();
    List<BulkItemRequest> shardRequests = requestsByShard.computeIfAbsent(shardId, shard -> new ArrayList<>());
    shardRequests.add(new BulkItemRequest(i, request));
}

if (requestsByShard.isEmpty()) {
    listener.onResponse(new BulkResponse(responses.toArray(new BulkItemResponse[responses.length()]), buildTookInMillis(startTimeNanos)));
    return;
}

final AtomicInteger counter = new AtomicInteger(requestsByShard.size());
// 当前节点ID
String nodeId = clusterService.localNode().getId();
for (Map.Entry<ShardId, List<BulkItemRequest>> entry : requestsByShard.entrySet()) {
    // 对每个分片
    final ShardId shardId = entry.getKey();
    final List<BulkItemRequest> requests = entry.getValue();
    // 创建该分片的批量操作请求
    BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, bulkRequest.getRefreshPolicy(),
            requests.toArray(new BulkItemRequest[requests.size()]));
    bulkShardRequest.waitForActiveShards(bulkRequest.waitForActiveShards());
    bulkShardRequest.timeout(bulkRequest.timeout());
    if (task != null) {
        bulkShardRequest.setParentTask(nodeId, task.getId());
    }
    // 执行该请求
    shardBulkAction.execute(bulkShardRequest, new ActionListener<BulkShardResponse>() {
        ...
}

// 找到主分片
final ShardRouting primary = primary(state);
if (retryIfUnavailable(state, primary)) {
    return;
}
final DiscoveryNode node = state.nodes().get(primary.currentNodeId());
// 主分片在当前节点就直接本地执行，否则就调用该远程节点执行
if (primary.currentNodeId().equals(state.nodes().getLocalNodeId())) {
    performLocalAction(state, primary, node, indexMetaData);
} else {
    performRemoteAction(state, primary, node);
}

public PrimaryResult perform(Request request) throws Exception {
    PrimaryResult result = shardOperationOnPrimary(request, indexShard);
    assert result.replicaRequest() == null || result.finalFailure == null : "a replica request [" + result.replicaRequest()
        + "] with a primary failure [" + result.finalFailure + "]";
    return result;
}

private static BulkItemResultHolder executeUpdateRequest(UpdateRequest updateRequest, IndexShard primary,
                                                         IndexMetaData metaData, BulkShardRequest request,
                                                         int requestIndex, UpdateHelper updateHelper,
                                                         LongSupplier nowInMillis,
                                                         final MappingUpdatePerformer mappingUpdater) throws Exception {
    BulkItemRequest primaryItemRequest = request.items()[requestIndex];
    assert primaryItemRequest.request() == updateRequest
            : "expected bulk item request to contain the original update request, got: " +
            primaryItemRequest.request() + " and " + updateRequest;

    BulkItemResultHolder holder = null;
    // There must be at least one attempt
    // 保证至少执行一次，因此重试
    int maxAttempts = Math.max(1, updateRequest.retryOnConflict());
    for (int attemptCount = 0; attemptCount < maxAttempts; attemptCount++) {

        holder = executeUpdateRequestOnce(updateRequest, primary, metaData, request.index(), updateHelper,
                nowInMillis, primaryItemRequest, request.items()[requestIndex].id(), mappingUpdater);

        // It was either a successful request, or it was a non-conflict failure
        if (holder.isVersionConflict() == false) {
            return holder;
        }
    }
    // We ran out of tries and haven't returned a valid bulk item response, so return the last one generated
    return holder;
}

public IndexResult index(Index index) throws IOException {
    assert Objects.equals(index.uid().field(), uidField) : index.uid().field();
    final boolean doThrottle = index.origin().isRecovery() == false;
    try (ReleasableLock releasableLock = readLock.acquire()) {
        ensureOpen();
        assert assertIncomingSequenceNumber(index.origin(), index.seqNo());
        assert assertVersionType(index);
        try (Releasable ignored = versionMap.acquireLock(index.uid().bytes());
            Releasable indexThrottle = doThrottle ? () -> {} : throttle.acquireThrottle()) {
            lastWriteNanos = index.startTime();
            /* A NOTE ABOUT APPEND ONLY OPTIMIZATIONS:
             * if we have an autoGeneratedID that comes into the engine we can potentially optimize
             * and just use addDocument instead of updateDocument and skip the entire version and index lookupVersion across the board.
             * Yet, we have to deal with multiple document delivery, for this we use a property of the document that is added
             * to detect if it has potentially been added before. We use the documents timestamp for this since it's something
             * that:
             *  - doesn't change per document
             *  - is preserved in the transaction log
             *  - and is assigned before we start to index / replicate
             * NOTE: it's not important for this timestamp to be consistent across nodes etc. it's just a number that is in the common
             * case increasing and can be used in the failure case when we retry and resent documents to establish a happens before relationship.
             * for instance:
             *  - doc A has autoGeneratedIdTimestamp = 10, isRetry = false
             *  - doc B has autoGeneratedIdTimestamp = 9, isRetry = false
             *
             *  while both docs are in in flight, we disconnect on one node, reconnect and send doc A again
             *  - now doc A' has autoGeneratedIdTimestamp = 10, isRetry = true
             *
             *  if A' arrives on the shard first we update maxUnsafeAutoIdTimestamp to 10 and use update document. All subsequent
             *  documents that arrive (A and B) will also use updateDocument since their timestamps are less than maxUnsafeAutoIdTimestamp.
             *  While this is not strictly needed for doc B it is just much simpler to implement since it will just de-optimize some doc in the worst case.
             *
             *  if A arrives on the shard first we use addDocument since maxUnsafeAutoIdTimestamp is < 10. A` will then just be skipped or calls
             *  updateDocument.
             */
            final IndexingStrategy plan;

            if (index.origin() == Operation.Origin.PRIMARY) {
                plan = planIndexingAsPrimary(index);
            } else {
                // non-primary mode (i.e., replica or recovery)
                plan = planIndexingAsNonPrimary(index);
            }

            final IndexResult indexResult;
            if (plan.earlyResultOnPreFlightError.isPresent()) {
                indexResult = plan.earlyResultOnPreFlightError.get();
                assert indexResult.getResultType() == Result.Type.FAILURE : indexResult.getResultType();
            } else if (plan.indexIntoLucene) {
                indexResult = indexIntoLucene(index, plan);
            } else {
                indexResult = new IndexResult(
                        plan.versionForIndexing, getPrimaryTerm(), plan.seqNoForIndexing, plan.currentNotFoundOrDeleted);
            }
            if (index.origin() != Operation.Origin.LOCAL_TRANSLOG_RECOVERY) {
                final Translog.Location location;
                if (indexResult.getResultType() == Result.Type.SUCCESS) {
                    location = translog.add(new Translog.Index(index, indexResult));
                } else if (indexResult.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
                    // if we have document failure, record it as a no-op in the translog with the generated seq_no
                    location = translog.add(new Translog.NoOp(indexResult.getSeqNo(), index.primaryTerm(), indexResult.getFailure().toString()));
                } else {
                    location = null;
                }
                indexResult.setTranslogLocation(location);
            }
            if (plan.indexIntoLucene && indexResult.getResultType() == Result.Type.SUCCESS) {
                final Translog.Location translogLocation = trackTranslogLocation.get() ? indexResult.getTranslogLocation() : null;
                versionMap.maybePutIndexUnderLock(index.uid().bytes(),
                    new IndexVersionValue(translogLocation, plan.versionForIndexing, plan.seqNoForIndexing, index.primaryTerm()));
            }
            if (indexResult.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
                localCheckpointTracker.markSeqNoAsCompleted(indexResult.getSeqNo());
            }
            indexResult.setTook(System.nanoTime() - index.startTime());
            indexResult.freeze();
            return indexResult;
        }
    } catch (RuntimeException | IOException e) {
        try {
            maybeFailEngine("index", e);
        } catch (Exception inner) {
            e.addSuppressed(inner);
        }
        throw e;
    }
}

private void performOnReplica(final ShardRouting shard, final ReplicaRequest replicaRequest, final long globalCheckpoint) {
    if (logger.isTraceEnabled()) {
        logger.trace("[{}] sending op [{}] to replica {} for request [{}]", shard.shardId(), opType, shard, replicaRequest);
    }

    totalShards.incrementAndGet();
    pendingActions.incrementAndGet();
    // 发HTTP请求给副分片
    replicasProxy.performOn(shard, replicaRequest, globalCheckpoint, new ActionListener<ReplicaResponse>() {
        ...
}

private AsyncSingleAction(Request request, ActionListener<Response> listener) {
    this.listener = listener;

    ClusterState clusterState = clusterService.state();
    if (logger.isTraceEnabled()) {
        logger.trace("executing [{}] based on cluster state version [{}]", request, clusterState.version());
    }
    // 集群nodes列表
    nodes = clusterState.nodes();
    ClusterBlockException blockException = checkGlobalBlock(clusterState);
    if (blockException != null) {
        throw blockException;
    }

    String concreteSingleIndex;
    if (resolveIndex(request)) {
        concreteSingleIndex = indexNameExpressionResolver.concreteSingleIndex(clusterState, request).getName();
    } else {
        concreteSingleIndex = request.index();
    }
    this.internalRequest = new InternalRequest(request, concreteSingleIndex);
    // 解析请求，更新自定义routing
    resolveRequest(clusterState, internalRequest);

    blockException = checkRequestBlock(clusterState, internalRequest);
    if (blockException != null) {
        throw blockException;
    }
    // 根据路由算法计算得到目的shard迭代器，或者根据优先级选择目标节点
    this.shardIt = shards(clusterState, internalRequest);
}

public <T extends TransportResponse> void sendRequest(final DiscoveryNode node, final String action,
                                                            final TransportRequest request,
                                                            final TransportResponseHandler<T> handler) {
    try {
        Transport.Connection connection = getConnection(node);
        sendRequest(connection, action, request, TransportRequestOptions.EMPTY, handler);
    } catch (NodeNotConnectedException ex) {
        // the caller might not handle this so we invoke the handler
        handler.handleException(ex);
    }
}

public Transport.Connection getConnection(DiscoveryNode node) {
    if (isLocalNode(node)) {
        return localNodeConnection;
    } else {
        return transport.getConnection(node);
    }
}

public void messageReceived(final Request request, final TransportChannel channel) throws Exception {
    if (logger.isTraceEnabled()) {
        logger.trace("executing [{}] on shard [{}]", request, request.internalShardId);
    }
    Response response = shardOperation(request, request.internalShardId);
    channel.sendResponse(response);
}

protected GetResponse shardOperation(GetRequest request, ShardId shardId) {
    IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
    IndexShard indexShard = indexService.getShard(shardId.id());

    if (request.refresh() && !request.realtime()) {
        indexShard.refresh("refresh_flag_get");
    }

    GetResult result = indexShard.getService().get(request.type(), request.id(), request.storedFields(),
            request.realtime(), request.version(), request.versionType(), request.fetchSourceContext());
    return new GetResponse(result);
}

private GetResult innerGet(String type, String id, String[] gFields, boolean realtime, long version, VersionType versionType,
                           FetchSourceContext fetchSourceContext, boolean readFromTranslog) {
    fetchSourceContext = normalizeFetchSourceContent(fetchSourceContext, gFields);
    final Collection<String> types;
    // 处理_all选项
    if (type == null || type.equals("_all")) {
        types = mapperService.types();
    } else {
        types = Collections.singleton(type);
    }

    Engine.GetResult get = null;
    for (String typeX : types) {
        Term uidTerm = mapperService.createUidTerm(typeX, id);
        if (uidTerm != null) {
            // 调用Engine读取数据
            get = indexShard.get(new Engine.Get(realtime, readFromTranslog, typeX, id, uidTerm)
                    .version(version).versionType(versionType));
            if (get.exists()) {
                type = typeX;
                break;
            } else {
                get.release();
            }
        }
    }

    ...

        // 过滤返回结果
        // 根据type、id、DocumentMapper等信息从刚刚获取的信息中获取数据，对指定的field、source进行过滤
        // 把结果存于GetResult返回
        return innerGetLoadFromStoredFields(type, id, gFields, fetchSourceContext, get, mapperService);
}

public Engine.GetResult get(Engine.Get get) {
    readAllowed();
    return getEngine().get(get, this::acquireSearcher);
}

public GetResult get(Get get, BiFunction<String, SearcherScope, Searcher> searcherFactory) throws EngineException {
    assert Objects.equals(get.uid().field(), uidField) : get.uid().field();
    try (ReleasableLock ignored = readLock.acquire()) {
        ensureOpen();
        SearcherScope scope;
        // 处理realtime选项，判断是否需要刷盘
        if (get.realtime()) {
            VersionValue versionValue = null;
            // versionMap写入索引的时候添加的，不会写入磁盘
            try (Releasable ignore = versionMap.acquireLock(get.uid().bytes())) {
                // we need to lock here to access the version map to do this truly in RT
                versionValue = getVersionFromMap(get.uid().bytes());
            }
            if (versionValue != null) {
                if (versionValue.isDelete()) {
                    return GetResult.NOT_EXISTS;
                }
                // 版本是否冲突
                if (get.versionType().isVersionConflictForReads(versionValue.version, get.version())) {
                    throw new VersionConflictEngineException(shardId, get.type(), get.id(),
                        get.versionType().explainConflictForReads(versionValue.version, get.version()));
                }
                if (get.isReadFromTranslog()) {
                    // this is only used for updates - API _GET calls will always read form a reader for consistency
                    // the update call doesn't need the consistency since it's source only + _parent but parent can go away in 7.0
                    if (versionValue.getLocation() != null) {
                        try {
                            Translog.Operation operation = translog.readOperation(versionValue.getLocation());
                            if (operation != null) {
                                // in the case of a already pruned translog generation we might get null here - yet very unlikely
                                TranslogLeafReader reader = new TranslogLeafReader((Translog.Index) operation, engineConfig
                                    .getIndexSettings().getIndexVersionCreated());
                                return new GetResult(new Searcher("realtime_get", new IndexSearcher(reader)),
                                    new VersionsAndSeqNoResolver.DocIdAndVersion(0, ((Translog.Index) operation).version(), reader, 0));
                            }
                        } catch (IOException e) {
                            maybeFailEngine("realtime_get", e); // lets check if the translog has failed with a tragic event
                            throw new EngineException(shardId, "failed to read operation from translog", e);
                        }
                    } else {
                        trackTranslogLocation.set(true);
                    }
                }
                // 执行刷盘操作
                refresh("realtime_get", SearcherScope.INTERNAL);
            }
            scope = SearcherScope.INTERNAL;
        } else {
            // we expose what has been externally expose in a point in time snapshot via an explicit refresh
            scope = SearcherScope.EXTERNAL;
        }
        // 调用Searcher读取数据
        // no version, get the version from the index, we know that we refresh on flush
        return getFromSearcher(get, searcherFactory, scope);
    }
}