stateDiagram-v2
    [*] --> CREATED: IndexService 创建
    CREATED --> RECOVERING: 分片恢复开始
    RECOVERING --> POST_RECOVERY: 恢复完成
    POST_RECOVERY --> STARTED: 分片启动
    STARTED --> RELOCATING: 分片迁移
    RELOCATING --> STARTED: 迁移完成
    STARTED --> CLOSED: 索引关闭
    CLOSED --> STARTED: 索引重新打开
    STARTED --> [*]: 索引删除
    CLOSED --> [*]: 索引删除

flowchart TB
    subgraph IndexModule["索引模块"]
        direction TB

        subgraph IndexService["IndexService 索引服务"]
            IS[IndexService<br/>索引实例管理]
            IM[IndexModule<br/>索引模块配置]
            IC[IndexCache<br/>缓存管理]
            IFD[IndexFieldDataService<br/>FieldData 服务]
        end

        subgraph Shards["分片管理"]
            Shard1[IndexShard<br/>主分片 0]
            Shard2[IndexShard<br/>副本分片 0]
            ShardN[IndexShard<br/>分片 N]
        end

        subgraph ShardComponents["分片组件"]
            Store[Store<br/>存储管理]
            Engine[Engine<br/>存储引擎]
            Translog[Translog<br/>事务日志]
            Mapper[MapperService<br/>映射服务]
            Codec[CodecService<br/>编解码服务]
        end

        subgraph Recovery["恢复机制"]
            RecTarget[RecoveryTarget<br/>恢复目标]
            RecSource[RecoverySource<br/>恢复源]
            RecState[RecoveryState<br/>恢复状态]
        end

        subgraph Operations["操作处理"]
            IndexOp[Index Operation<br/>索引操作]
            DeleteOp[Delete Operation<br/>删除操作]
            GetOp[Get Operation<br/>查询操作]
            BulkOp[Bulk Operation<br/>批量操作]
        end
    end

    subgraph External["外部依赖"]
        Lucene[Apache Lucene<br/>全文检索库]
        ClusterSvc[Cluster Service<br/>集群服务]
        TransportSvc[Transport Service<br/>传输服务]
    end

    IS --> IM
    IS --> IC
    IS --> IFD
    IS --> Shard1
    IS --> Shard2
    IS --> ShardN

    Shard1 --> Store
    Shard1 --> Engine
    Shard1 --> Translog
    Shard1 --> Mapper
    Shard1 --> Codec

    Shard1 --> RecTarget
    Shard2 --> RecSource
    RecTarget --> RecState

    IndexOp --> Shard1
    DeleteOp --> Shard1
    GetOp --> Shard1
    BulkOp --> Shard1

    Engine --> Lucene
    IS --> ClusterSvc
    Shard1 --> TransportSvc

    style IndexModule fill:#E1F5E1
    style IndexService fill:#FFF4E1
    style Shards fill:#E1E1F5
    style ShardComponents fill:#FFE1E1
    style Recovery fill:#F5E1FF
    style Operations fill:#E1F5FF

flowchart LR
    A[Index Request] --> B{路由计算}
    B --> C[主分片]
    C --> D[解析文档]
    D --> E{Mapping 更新?}
    E -->|是| F[更新 Mapping]
    E -->|否| G[执行索引]
    F --> G
    G --> H[写入 Engine]
    H --> I[更新 VersionMap]
    I --> J[写入 Lucene]
    J --> K[写入 Translog]
    K --> L[复制到副本]
    L --> M[返回响应]

flowchart TB
    A[分片分配] --> B{恢复类型}
    B -->|EmptyStore| C[创建空分片]
    B -->|ExistingStore| D[从本地恢复]
    B -->|Peer| E[从副本恢复]
    B -->|Snapshot| F[从快照恢复]

    C --> G[Translog 回放]
    D --> G
    E --> H[阶段1: 文件复制]
    F --> I[从仓库下载]

    H --> J[阶段2: Translog 复制]
    I --> G
    J --> G

    G --> K[标记为 Started]

classDiagram
    class IndexService {
        -IndexSettings indexSettings
        -MapperService mapperService
        -IndexCache indexCache
        -Map~Integer,IndexShard~ shards
        +createShard(routing) IndexShard
        +removeShard(shardId, reason)
        +getShard(shardId) IndexShard
        +refresh(source)
        +flush(request)
    }

    class IndexShard {
        -ShardId shardId
        -ShardRouting shardRouting
        -IndexShardState state
        -AtomicReference~Engine~ currentEngineRef
        -Store store
        -MapperService mapperService
        -ReplicationTracker replicationTracker
        +applyIndexOperationOnPrimary() IndexResult
        +applyIndexOperationOnReplica() IndexResult
        +applyDeleteOperationOnPrimary() DeleteResult
        +get(get) GetResult
        +refresh(source)
        +flush(request)
    }

    class Store {
        -Directory directory
        -ShardLock shardLock
        -IndexSettings indexSettings
        +getMetadata(commit) MetadataSnapshot
        +verify()
        +cleanupFiles(segmentInfos)
    }

    class Engine {
        <<interface>>
        +index(index) IndexResult
        +delete(delete) DeleteResult
        +get(get) GetResult
        +refresh(source)
        +flush(force, waitIfOngoing) FlushResult
        +acquireSearcher(source) Searcher
        +getTranslog() Translog
    }

    class InternalEngine {
        -LiveVersionMap versionMap
        -IndexWriter indexWriter
        -Translog translog
        -LocalCheckpointTracker localCheckpointTracker
        -EngineConfig config
        -AtomicLong lastRefreshTime
        +index(index) IndexResult
        +delete(delete) DeleteResult
        +refresh(source)
    }

    class ShardRouting {
        -ShardId shardId
        -String currentNodeId
        -boolean primary
        -ShardRoutingState state
        -RecoverySource recoverySource
        +active() boolean
        +assignedToNode() boolean
    }

    IndexService "1" --> "*" IndexShard : manages
    IndexShard "1" --> "1" Store : uses
    IndexShard "1" --> "1" Engine : uses
    Engine <|.. InternalEngine : implements
    IndexShard "1" --> "1" ShardRouting : routing info

classDiagram
    class LiveVersionMap {
        -KeyedLock~BytesRef~ keyedLock
        -Maps maps
        -Map~BytesRef,DeleteVersionValue~ tombstones
        +getUnderLock(uid) VersionValue
        +putIndexUnderLock(uid, version)
        +putDeleteUnderLock(uid, version)
        +acquireLock(uid) Releasable
        +releaseLock(uid)
    }

    class VersionValue {
        <<abstract>>
        -long version
        -long seqNo
        -long term
        +isDelete() boolean
        +ramBytesUsed() long
    }

    class IndexVersionValue {
        +isDelete() false
    }

    class DeleteVersionValue {
        -long time
        +isDelete() true
    }

    class ReplicationTracker {
        -String shardAllocationId
        -long operationPrimaryTerm
        -long globalCheckpoint
        -Map~String,CheckpointState~ checkpoints
        -ReplicationGroup replicationGroup
        +getGlobalCheckpoint() long
        +updateFromMaster(shardRoutingTable)
        +initiateTracking(allocationId)
        +markAllocationIdAsInSync(allocationId)
    }

    class CheckpointState {
        -long localCheckpoint
        -long globalCheckpoint
        -boolean inSync
        -boolean tracked
        +getLocalCheckpoint() long
        +getGlobalCheckpoint() long
    }

    class LocalCheckpointTracker {
        -long checkpoint
        -LongObjectHashMap~CountedBitSet~ processedSeqNo
        -long maxSeqNo
        +generateSeqNo() long
        +markSeqNoAsProcessed(seqNo)
        +getProcessedCheckpoint() long
        +getMaxSeqNo() long
    }

    LiveVersionMap "1" --> "*" VersionValue : stores
    VersionValue <|-- IndexVersionValue
    VersionValue <|-- DeleteVersionValue
    ReplicationTracker "1" --> "*" CheckpointState : tracks replicas
    ReplicationTracker "1" --> "1" LocalCheckpointTracker : local checkpoint

classDiagram
    class EngineConfig {
        -ShardId shardId
        -IndexSettings indexSettings
        -Analyzer analyzer
        -Similarity similarity
        -CodecService codecService
        -TranslogConfig translogConfig
        -MergePolicy mergePolicy
        -MergeScheduler mergeScheduler
        -IndexWriterConfig indexWriterConfig
    }

    class Translog {
        -TranslogConfig config
        -Path location
        -List~BaseTranslogReader~ readers
        -TranslogWriter current
        -AtomicLong globalCheckpoint
        +add(operation) Location
        +readLocation(location) Operation
        +newSnapshot() Snapshot
        +rollGeneration()
        +trimUnreferencedReaders()
    }

    class TranslogWriter {
        -FileChannel channel
        -AtomicLong lastSyncedOffset
        -Checkpoint checkpoint
        +add(data) Location
        +sync()
        +closeIntoReader() TranslogReader
    }

    class Checkpoint {
        -long offset
        -int numOps
        -long generation
        -long minSeqNo
        -long maxSeqNo
        -long globalCheckpoint
        -long minTranslogGeneration
        -long trimmedAboveSeqNo
    }

    class Segment {
        -String name
        -long generation
        -boolean committed
        -boolean search
        -long sizeInBytes
        -int docCount
        -int delDocCount
        -Version version
        -Codec codec
        +ramBytesUsed() long
    }

    InternalEngine "1" --> "1" EngineConfig : config
    InternalEngine "1" --> "1" Translog : translog
    Translog "1" --> "1" TranslogWriter : current writer
    Translog "1" --> "*" Checkpoint : checkpoints
    InternalEngine "1" --> "*" Segment : segments

// IndexService 管理单个索引的所有分片
public class IndexService extends AbstractIndexComponent
    implements IndicesClusterStateService.AllocatedIndex<IndexShard> {

    // 索引元数据和配置
    private final IndexSettings indexSettings;
    private final MapperService mapperService;
    private final IndexCache indexCache;
    private final IndexFieldDataService indexFieldData;

    // 分片映射 shardId -> IndexShard
    private volatile Map<Integer, IndexShard> shards;

    // 创建新分片
    public synchronized IndexShard createShard(
        ShardRouting routing,
        GlobalCheckpointSyncer globalCheckpointSyncer,
        RetentionLeaseSyncer retentionLeaseSyncer
    ) throws IOException;

    // 移除分片
    public synchronized void removeShard(int shardId, String reason);

    // 获取分片
    public IndexShard getShard(int shardId);
}

// IndexShard 表示单个分片,是索引模块的核心类
public class IndexShard extends AbstractIndexShardComponent
    implements IndicesClusterStateService.Shard {

    // 分片路由信息
    protected volatile ShardRouting shardRouting;

    // 分片状态: CREATED/RECOVERING/POST_RECOVERY/STARTED/CLOSED
    protected volatile IndexShardState state;

    // 存储引擎引用
    private final AtomicReference<Engine> currentEngine;

    // 存储管理
    private final Store store;

    // 事务日志
    private final TranslogConfig translogConfig;

    // 索引操作 - 主分片
    public Engine.IndexResult applyIndexOperationOnPrimary(
        long version,
        VersionType versionType,
        SourceToParse sourceToParse,
        long ifSeqNo,
        long ifPrimaryTerm,
        long autoGeneratedTimestamp,
        boolean isRetry
    ) throws IOException;

    // 索引操作 - 副本分片
    public Engine.IndexResult applyIndexOperationOnReplica(
        long seqNo,
        long opPrimaryTerm,
        long version,
        long autoGeneratedTimeStamp,
        boolean isRetry,
        SourceToParse sourceToParse
    ) throws IOException;

    // 删除操作
    public Engine.DeleteResult applyDeleteOperationOnPrimary(
        long version,
        String id,
        VersionType versionType,
        long ifSeqNo,
        long ifPrimaryTerm
    ) throws IOException;

    // 获取文档
    public Engine.GetResult get(Engine.Get get);

    // Refresh 操作
    public void refresh(String source);

    // Flush 操作
    public void flush(FlushRequest request);

    // 强制合并
    public void forceMerge(ForceMergeRequest request);

    // 恢复分片
    public void startRecovery(
        RecoveryState recoveryState,
        PeerRecoveryTargetService.RecoveryListener recoveryListener,
        RepositoriesService repositoriesService
    );
}

// Store 管理分片的 Lucene Directory 和文件
public class Store extends AbstractIndexShardComponent implements Closeable, RefCounted {

    // Lucene Directory
    private final Directory directory;

    // 分片锁,防止多个进程同时访问
    private final ShardLock shardLock;

    // 获取元数据快照
    public MetadataSnapshot getMetadata(IndexCommit commit) throws IOException;

    // 验证索引完整性
    public void verify() throws IOException;

    // 检查索引
    public CheckIndex.Status checkIndex(PrintStream printStream) throws IOException;

    // 清理文件
    public void cleanupFiles(SegmentInfos segmentInfos) throws IOException;
}

// Engine 接口,定义存储引擎的核心操作
public interface Engine extends Closeable {

    // 索引文档
    IndexResult index(Index index) throws IOException;

    // 删除文档
    DeleteResult delete(Delete delete) throws IOException;

    // 获取文档
    GetResult get(Get get, BiFunction<String, SearcherScope, Searcher> searcherFactory);

    // Refresh,使文档可搜索
    void refresh(String source) throws EngineException;

    // Flush,持久化到磁盘
    FlushResult flush(boolean force, boolean waitIfOngoing) throws EngineException;

    // 强制合并 Segment
    void forceMerge(
        boolean flush,
        int maxNumSegments,
        boolean onlyExpungeDeletes,
        String forceMergeUUID
    ) throws IOException;

    // 获取 Searcher
    Searcher acquireSearcher(String source, SearcherScope scope);

    // 获取 Translog
    Translog getTranslog();
}

// IndexRequest - 索引文档请求
public class IndexRequest extends ReplicatedWriteRequest<IndexRequest>
    implements DocWriteRequest<IndexRequest> {

    // 文档 ID,不设置则自动生成
    private String id;

    // 路由键,用于计算文档所在分片，默认使用 id 作为路由键
    private String routing;

    // 文档源内容
    private final IndexSource indexSource;

    // 操作类型: INDEX(插入或更新) 或 CREATE(仅插入)
    private OpType opType = OpType.INDEX;

    // 版本号,用于乐观并发控制
    private long version = Versions.MATCH_ANY;
    private VersionType versionType = VersionType.INTERNAL;

    // 序列号并发控制(推荐方式)
    private long ifSeqNo = UNASSIGNED_SEQ_NO;
    private long ifPrimaryTerm = UNASSIGNED_PRIMARY_TERM;

    // Ingest Pipeline
    private String pipeline;

    // 是否必须是别名
    private boolean requireAlias;

    // 自动生成时间戳(用于 ID 冲突检测)
    private long autoGeneratedTimestamp = UNSET_AUTO_GENERATED_TIMESTAMP;

    // 是否为重试请求
    private boolean isRetry = false;

    // 动态模板映射
    private Map<String, String> dynamicTemplates = Map.of();
}

// IndexResponse - 索引文档响应
public class IndexResponse extends DocWriteResponse implements StatusToXContentObject {

    // 文档 ID
    private String _id;

    // 文档版本号(每次更新递增)
    private long _version;

    // 序列号(全局唯一,单调递增)
    private long _seq_no;

    // 主分片任期号
    private long _primary_term;

    // 操作结果: created, updated, deleted, not_found, noop
    private Result result;

    // 分片副本确认信息
    private ReplicationResponse.ShardInfo _shards;
}

// TransportIndexAction - 索引操作的传输层动作
// 实际委托给 TransportBulkAction 处理单个文档
public class TransportIndexAction extends TransportSingleItemBulkWriteAction<IndexRequest, DocWriteResponse> {

    public static final String NAME = "indices:data/write/index";

    @Inject
    public TransportIndexAction(
        ActionFilters actionFilters,
        TransportService transportService,
        TransportBulkAction bulkAction
    ) {
        // 将单文档请求转换为 Bulk 请求处理
        super(NAME, transportService, actionFilters, IndexRequest::new, bulkAction);
    }
}

// TransportBulkAction - 批量操作核心处理类
protected void doExecute(Task task, BulkRequest bulkRequest, ActionListener<BulkResponse> listener) {
    // 1. 解析请求,获取所有待处理的文档操作
    final long startTime = relativeTime();
    final AtomicArray<BulkItemResponse> responses = new AtomicArray<>(bulkRequest.requests.size());

    // 2. 处理 Ingest Pipeline(如果配置)
    if (clusterService.localNode().isIngestNode()) {
        processBulkIndexIngestRequest(task, bulkRequest, executorName, listener);
    } else {
        executeBulk(task, bulkRequest, startTime, listener, responses);
    }
}

private void executeBulk(
    Task task,
    BulkRequest bulkRequest,
    long startTimeNanos,
    ActionListener<BulkResponse> listener,
    AtomicArray<BulkItemResponse> responses
) {
    // 1. 创建自动索引(如果不存在且允许自动创建)
    // 2. 按分片分组请求
    final Map<ShardId, List<BulkItemRequest>> requestsByShard = new HashMap<>();

    for (int i = 0; i < bulkRequest.requests.size(); i++) {
        DocWriteRequest<?> request = bulkRequest.requests.get(i);
        // 解析索引名(可能是别名/Data Stream)
        // 计算路由,确定目标分片
        ShardId shardId = clusterService.operationRouting().indexShards(
            clusterState,
            request.index(),
            request.id(),
            request.routing()
        ).shardId();

        requestsByShard.computeIfAbsent(shardId, k -> new ArrayList<>())
            .add(new BulkItemRequest(i, request));
    }

    // 3. 并发执行所有分片的批量操作
    if (requestsByShard.isEmpty()) {
        listener.onResponse(new BulkResponse(responses.toArray(new BulkItemResponse[responses.length()]), buildTookInMillis(startTimeNanos)));
        return;
    }

    final AtomicInteger counter = new AtomicInteger(requestsByShard.size());
    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[0])
        );

        // 执行分片级别的批量操作
        shardBulkAction.execute(bulkShardRequest, new ActionListener<BulkShardResponse>() {
            @Override
            public void onResponse(BulkShardResponse bulkShardResponse) {
                // 收集响应
                for (BulkItemResponse itemResponse : bulkShardResponse.getResponses()) {
                    responses.set(itemResponse.getItemId(), itemResponse);
                }

                // 所有分片完成后返回
                if (counter.decrementAndGet() == 0) {
                    finalizeBulkRequest(task, bulkRequest, responses, startTimeNanos, listener);
                }
            }

            @Override
            public void onFailure(Exception e) {
                // 记录分片级别失败
                for (BulkItemRequest request : requests) {
                    responses.set(request.id(), new BulkItemResponse(request.id(),
                        request.request().opType(),
                        new BulkItemResponse.Failure(request.index(), request.id(), e)));
                }

                if (counter.decrementAndGet() == 0) {
                    finalizeBulkRequest(task, bulkRequest, responses, startTimeNanos, listener);
                }
            }
        });
    }
}

// TransportShardBulkAction - 分片级别批量操作
protected void shardOperationOnPrimary(
    BulkShardRequest request,
    IndexShard primary,
    ActionListener<PrimaryResult<BulkShardRequest, BulkShardResponse>> listener
) {
    // 在主分片上执行批量操作
    Executor executor = threadPool.executor(ThreadPool.Names.WRITE);
    executor.execute(new AbstractRunnable() {
        @Override
        protected void doRun() {
            // 1. 获取 Engine
            final Engine.IndexingMemoryController memoryController =
                indicesService.indexingMemoryController();

            // 2. 逐个处理文档操作
            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();

                    try {
                        // 调用 IndexShard 执行索引操作
                        Engine.IndexResult result = primary.applyIndexOperationOnPrimary(
                            indexRequest.version(),
                            indexRequest.versionType(),
                            new SourceToParse(
                                indexRequest.index(),
                                indexRequest.id(),
                                indexRequest.source(),
                                indexRequest.getContentType(),
                                indexRequest.routing()
                            ),
                            indexRequest.ifSeqNo(),
                            indexRequest.ifPrimaryTerm(),
                            indexRequest.getAutoGeneratedTimestamp(),
                            indexRequest.isRetry()
                        );

                        // 构建响应
                        if (result.getResultType() == Engine.Result.Type.SUCCESS) {
                            IndexResponse response = new IndexResponse(
                                primary.shardId(),
                                indexRequest.id(),
                                result.getSeqNo(),
                                result.getTerm(),
                                result.getVersion(),
                                result.isCreated()
                            );
                            responses[i] = new BulkItemResponse(i, indexRequest.opType(), response);
                        } else {
                            // 处理失败情况
                            responses[i] = new BulkItemResponse(i, indexRequest.opType(),
                                new BulkItemResponse.Failure(
                                    indexRequest.index(),
                                    indexRequest.id(),
                                    result.getFailure()
                                )
                            );
                        }
                    } catch (Exception e) {
                        responses[i] = new BulkItemResponse(i, indexRequest.opType(),
                            new BulkItemResponse.Failure(
                                indexRequest.index(),
                                indexRequest.id(),
                                e
                            )
                        );
                    }
                }
            }

            // 3. 返回主分片处理结果
            BulkShardResponse response = new BulkShardResponse(request.shardId(), responses);
            listener.onResponse(new PrimaryResult<>(request, response));
        }
    });
}

# 1. 获取文档当前版本
GET /products/_doc/1

# Response:
# {
#   "_seq_no": 5,
#   "_primary_term": 1,
#   ...
# }

# 2. 基于获取到的版本更新
PUT /products/_doc/1?if_seq_no=5&if_primary_term=1
{
  "name": "Updated Product",
  "price": 199.99
}

# 批量索引(推荐)
POST /_bulk
{"index":{"_index":"products","_id":"1"}}
{"name":"Product 1","price":99.99}
{"index":{"_index":"products","_id":"2"}}
{"name":"Product 2","price":199.99}

# 单个索引(不推荐)
PUT /products/_doc/1
{"name":"Product 1","price":99.99}
PUT /products/_doc/2
{"name":"Product 2","price":199.99}

# 实时可见(性能最差)
PUT /products/_doc/1?refresh=true
{"name":"Product"}

# 等待刷新完成(性能较差)
PUT /products/_doc/1?refresh=wait_for
{"name":"Product"}

# 不刷新(性能最好,默认)
PUT /products/_doc/1
{"name":"Product"}

# 将同一用户的文档路由到同一分片
PUT /orders/_doc/order1?routing=user123
{"user_id":"user123","product":"laptop"}

PUT /orders/_doc/order2?routing=user123
{"user_id":"user123","product":"mouse"}

# 搜索时指定路由,只查询一个分片
GET /orders/_search?routing=user123
{
  "query": {"term": {"user_id": "user123"}}
}

# 禁用 _source 存储(不需要返回原始文档时)
PUT /logs
{
  "mappings": {
    "_source": {"enabled": false}
  }
}

# 或者仅存储部分字段
PUT /logs
{
  "mappings": {
    "_source": {
      "includes": ["timestamp", "message"],
      "excludes": ["large_field"]
    }
  }
}

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant Primary as IndexShard (Primary)
    participant Mapper as MapperService
    participant Engine as InternalEngine
    participant VersionMap as LiveVersionMap
    participant Checkpoint as LocalCheckpointTracker
    participant Lucene as Lucene IndexWriter
    participant Translog as Translog
    participant Replicas as Replica Shards

    Client->>Primary: IndexRequest(doc, id)

    Primary->>Primary: 1. checkWriteAllowed()<br/>检查分片状态

    Primary->>Mapper: 2. parse(source)<br/>解析文档
    Mapper->>Mapper: 解析 JSON,构建 Lucene Document
    Mapper->>Mapper: 检查 Mapping 兼容性

    alt Mapping 需要更新
        Mapper-->>Primary: 返回 DynamicMappingsUpdate
        Primary-->>Client: 需要更新 Mapping<br/>(由上层处理)
    else Mapping 兼容
        Mapper-->>Primary: ParsedDocument
    end

    Primary->>Engine: 3. index(operation)

    Engine->>VersionMap: 4. acquireLock(uid)<br/>获取文档锁
    VersionMap-->>Engine: Releasable lock

    Engine->>VersionMap: 5. getUnderLock(uid)<br/>获取当前版本
    VersionMap-->>Engine: VersionValue (if exists)

    alt 版本冲突
        Engine->>Engine: 检查 ifSeqNo/ifPrimaryTerm
        Engine-->>Primary: VersionConflictException
        Primary-->>Client: 409 Conflict
    else 无冲突
        Engine->>Checkpoint: 6. generateSeqNo()<br/>生成序列号
        Checkpoint-->>Engine: seqNo=N

        Engine->>VersionMap: 7. putIndexUnderLock(uid, version)
        VersionMap->>VersionMap: 更新 current map

        alt 文档不存在
            Engine->>Lucene: 8a. addDocument(doc)
        else 文档已存在
            Engine->>Lucene: 8b. updateDocument(uid, doc)
        end
        Lucene-->>Engine: success

        Engine->>Translog: 9. add(operation, seqNo)
        Translog->>Translog: 序列化操作
        Translog->>Translog: 写入文件

        alt Durability = REQUEST
            Translog->>Translog: fsync()
        end
        Translog-->>Engine: Location

        Engine->>Checkpoint: 10. markSeqNoAsProcessed(seqNo)
        Checkpoint->>Checkpoint: 更新 checkpoint
        Checkpoint-->>Engine: success

        Engine->>VersionMap: 11. releaseLock(uid)

        Engine-->>Primary: IndexResult<br/>(created/updated, seqNo, version)
    end

    Primary->>Replicas: 12. ReplicationRequest<br/>(doc, seqNo, primaryTerm)

    loop 每个副本
        Replicas->>Replicas: applyIndexOperationOnReplica()
        Replicas->>Engine: index(doc, seqNo)
        Engine->>Lucene: addDocument/updateDocument
        Engine->>Translog: add(operation)
        Replicas-->>Primary: ACK
    end

    Primary->>Primary: 13. waitForActiveShards()<br/>等待多数副本确认

    Primary-->>Client: IndexResponse<br/>(id, version, seqNo, result)

sequenceDiagram
    autonumber
    participant Master as Master Node
    participant Primary as Primary Shard
    participant Replica as Replica Shard (Target)
    participant Store as Store (Target)
    participant Engine as Engine (Target)
    participant Translog as Translog (Target)

    Master->>Replica: 分配副本分片到节点
    Replica->>Replica: 创建 IndexShard<br/>state = RECOVERING

    Replica->>Primary: StartRecoveryRequest
    Primary->>Primary: 1. prepareForTranslogOperations()<br/>准备发送 Translog

    Primary->>Primary: 2. 获取 Retention Lease<br/>防止 Translog 被清理

    Note over Primary,Replica: Phase 1: 文件复制(Segment files)

    Primary->>Primary: 3. 创建 Lucene snapshot<br/>获取当前 Segment 列表
    Primary->>Replica: FileChunkRequest<br/>(segment files metadata)

    loop 每个缺失的 Segment 文件
        Primary->>Replica: FileChunkRequest<br/>(file data, offset, length)
        Replica->>Store: writeFileChunk(data)
        Store-->>Replica: success
    end

    Primary-->>Replica: Phase 1 完成<br/>(seqNo snapshot)

    Note over Primary,Replica: Phase 2: Translog 复制

    Primary->>Primary: 4. 获取 Translog snapshot<br/>(startSeqNo = recovery start)

    loop 批量发送 Translog 操作
        Primary->>Replica: TranslogOperationsRequest<br/>(operations[], maxSeqNo)

        loop 每个操作
            Replica->>Engine: applyIndexOperationOnReplica(op)
            Engine->>Engine: 按 seqNo 顺序应用
            Engine-->>Replica: success
        end

        Replica-->>Primary: ACK (localCheckpoint)
    end

    Primary->>Replica: FinalizeRecoveryRequest<br/>(globalCheckpoint)

    Replica->>Replica: 5. finalizeRecovery()
    Replica->>Engine: updateGlobalCheckpointOnReplica()
    Replica->>Translog: trimUnreferencedReaders()<br/>清理旧 Translog

    Replica->>Replica: 6. changeState(RECOVERING → POST_RECOVERY)
    Replica->>Replica: 7. changeState(POST_RECOVERY → STARTED)

    Replica->>Primary: RecoveryResponse<br/>(success, localCheckpoint)

    Primary->>Primary: 8. markAllocationIdAsInSync()<br/>标记副本为 in-sync

    Primary-->>Master: 副本恢复完成

INITIALIZING → RECOVERING → POST_RECOVERY → STARTED

sequenceDiagram
    autonumber
    participant Scheduler as Refresh Scheduler
    participant Shard as IndexShard
    participant Engine as InternalEngine
    participant Lucene as IndexWriter
    participant VersionMap as LiveVersionMap
    participant Searcher as SearcherManager

    Scheduler->>Shard: refresh(source="scheduled")

    Shard->>Engine: refresh(source, scope)

    Engine->>Engine: 1. 检查是否需要 Refresh<br/>(lastRefreshTime, dirty flag)

    alt 无新写入
        Engine-->>Shard: 跳过 Refresh
    else 有新写入
        Engine->>Lucene: 2. IndexWriter.flush()<br/>(不 commit,仅生成 Segment)

        Lucene->>Lucene: 将 IndexBuffer 写入新 Segment
        Lucene->>Lucene: 生成 .cfs/.cfe/.si 文件
        Lucene-->>Engine: success

        Engine->>Searcher: 3. SearcherManager.maybeRefresh()
        Searcher->>Searcher: 打开新的 IndexReader
        Searcher->>Searcher: 触发 RefreshListener
        Searcher-->>Engine: 新 Searcher

        Engine->>VersionMap: 4. beforeRefresh()
        VersionMap->>VersionMap: current → old<br/>创建新的 empty current

        Engine->>Engine: 5. 更新 lastRefreshTime

        Engine->>VersionMap: 6. afterRefresh()
        VersionMap->>VersionMap: 清空 old map<br/>释放内存

        Engine-->>Shard: RefreshResult
    end

    Shard->>Shard: 7. 触发 RefreshListener<br/>(通知等待 Refresh 的请求)

    Shard-->>Scheduler: success

# 分片和副本配置
index.number_of_shards: 5              # 主分片数,创建后不可修改
index.number_of_replicas: 1            # 副本数,可动态修改

# Refresh 配置
index.refresh_interval: 1s             # Refresh 间隔,默认 1s
index.max_refresh_listeners: 1000      # 最大 Refresh 监听器数量

# Translog 配置
index.translog.durability: request     # request: 每次请求 fsync, async: 异步 fsync
index.translog.sync_interval: 5s       # 异步 fsync 间隔
index.translog.flush_threshold_size: 512mb  # Translog 大小阈值,触发 Flush

# Merge 配置
index.merge.scheduler.max_thread_count: 1   # 合并线程数
index.merge.policy.max_merged_segment: 5gb  # 最大 Segment 大小

# 缓存配置
index.queries.cache.enabled: true      # 查询缓存开关
index.requests.cache.enable: true      # 请求缓存开关

# 存储配置
index.store.type: fs                   # 存储类型: fs, niofs, mmapfs
index.codec: default                   # 编解码器: default, best_compression

PUT /my-index/_settings
{
  "index.refresh_interval": "30s"
}

PUT /my-index/_settings
{
  "index.translog.durability": "async",
  "index.translog.sync_interval": "5s"
}

PUT /my-index/_settings
{
  "index.number_of_replicas": 0
}

indices.memory.index_buffer_size: 20%

PUT /my-index/_settings
{
  "index.queries.cache.enabled": true
}

{
  "query": {
    "bool": {
      "filter": [
        { "term": { "status": "published" } }
      ]
    }
  }
}

POST /my-index/_forcemerge?max_num_segments=1

PUT /my-index/_settings
{
  "index.codec": "best_compression"
}

PUT /my-index
{
  "mappings": {
    "_source": {
      "enabled": false
    }
  }
}

# 索引统计
GET /_stats/indexing,search,store,merge,refresh

# 分片信息
GET /_cat/shards?v&h=index,shard,prirep,state,docs,store,node

# 索引健康
GET /_cluster/health?level=indices

# 恢复状态
GET /_cat/recovery?v&active_only=true