源码精读：WorldManagerService — 数据世界的 Git 操作

“

系列：S9 源码精读 · 第 3 篇 | 难度：高级 | 阅读时间：25 分钟

#TL;DR

WorldManagerService 是 Control Layer 中管理"数据世界"（World）的核心服务，通过集成 Nessie 实现了类 Git 的数据版本管理。每个 World 是一个完全隔离的数据空间，支持分支、合并、发布（Tag）、时间旅行和反事实分析。本文将深入分析 Nessie 集成架构、World 生命周期管理、BranchManager 的分支操作、MergeService 的冲突检测与三路合并策略、CounterfactualBranchService 的假设分析实现，以及配额管理机制。

#1. World 概念与架构定位

在 coomia-dip 中，World 是最顶层的数据隔离单元。它对标 Palantir Foundry 中的 Namespace/Project 概念，但增加了 Git 语义：

Code

World "production"
├── Branch "main"           # 主分支（生产数据）
├── Branch "dev-feature-x"  # 开发分支
├── Branch "whatif-budget"   # 反事实分析分支
├── Tag "v2025.03.01"       # Release 快照
└── Tag "v2025.02.15"       # 历史 Release

每个 World 在 Nessie 中对应一个 Repository 的隔离空间。Nessie 是 coomia-dip 的"数据目录版本控制系统"——它不存储数据本身（数据在 Doris/Iceberg 中），而是管理数据的元数据版本（哪些表存在、Schema 是什么、每条记录的版本快照 ID）。

#2. 服务类结构：六大协作者

Java

@GrpcService
@Slf4j
@RequiredArgsConstructor
public class WorldManagerServiceImpl
        extends WorldManagerServiceGrpc.WorldManagerServiceImplBase {

    private final WorldService worldService;
    private final BranchManager branchManager;
    private final MergeService mergeService;
    private final NessieClientWrapper nessieClient;
    private final CounterfactualBranchService counterfactualBranchService;
    private final WorldMapper worldMapper;
}

协作者	职责
`WorldService`	World CRUD、状态管理
`BranchManager`	分支创建/删除/列表/Commit 日志
`MergeService`	分支合并、冲突检测
`NessieClientWrapper`	Nessie REST API 封装
`CounterfactualBranchService`	反事实（What-If）分析分支
`WorldMapper`	Proto ↔ Domain 映射

设计意图：将复杂的 World 管理拆分为多个单一职责的服务，而非一个巨大的 God Class。WorldManagerServiceImpl 仅负责 gRPC 协议适配和请求路由。

#3. World 生命周期管理

Java

@Override
public void createWorld(CreateWorldRequest request,
        StreamObserver<WorldResponse> responseObserver) {
    try {
        log.info("CreateWorld: name={}", request.getWorldName());

        // 1. 配额检查
        worldService.checkQuota(
            WorldContextHolder.getTenantId());

        // 2. 创建 World 记录
        World world = World.builder()
            .worldId(UUID.randomUUID().toString())
            .worldName(request.getWorldName())
            .description(request.getDescription())
            .status(WorldStatus.CREATING)
            .tenantId(WorldContextHolder.getTenantId())
            .createdBy(WorldContextHolder.getUserId())
            .build();

        World created = worldService.create(world);

        // 3. 在 Nessie 中创建对应分支
        nessieClient.createBranch(
            created.getWorldId(),
            "main",          // 默认主分支
            null             // 从空状态开始
        );

        // 4. 更新状态为 ACTIVE
        created.setStatus(WorldStatus.ACTIVE);
        worldService.update(created);

        responseObserver.onNext(
            worldMapper.toResponse(created));
        responseObserver.onCompleted();
    } catch (Exception e) {
        handleException(e, responseObserver);
    }
}

创建流程分两阶段：先创建 World 记录（状态为 CREATING），再在 Nessie 中创建分支。如果 Nessie 调用失败，World 记录保持 CREATING 状态，后台任务会定期清理这些"僵尸"World。

Java

public enum WorldStatus {
    CREATING,     // Nessie 分支创建中
    ACTIVE,       // 正常使用
    SUSPENDED,    // 暂停（超配额等）
    DELETING,     // 删除中
    DELETED       // 已删除
}

#4. Nessie 集成：NessieClientWrapper

Java

@Component
public class NessieClientWrapper {

    private final NessieApiV2 nessieApi;

    public NessieClientWrapper(NessieProperties properties) {
        this.nessieApi = HttpClientBuilder
            .builder()
            .withUri(URI.create(properties.getUri()))
            .withAuthentication(
                BearerAuthenticationProvider.create(
                    properties.getToken()))
            .build(NessieApiV2.class);
    }

    public Branch createBranch(String worldId,
            String branchName, String fromHash) {
        String fullBranchName = formatBranchName(
            worldId, branchName);
        try {
            return nessieApi.createReference()
                .reference(Branch.of(fullBranchName, fromHash))
                .create();
        } catch (NessieConflictException e) {
            throw new BranchAlreadyExistsException(
                branchName, worldId);
        } catch (NessieNotFoundException e) {
            throw new WorldNotFoundException(worldId);
        }
    }

    public LogResponse getCommitLog(String worldId,
            String branchName, int maxEntries) {
        String fullBranchName = formatBranchName(
            worldId, branchName);
        return nessieApi.getCommitLog()
            .refName(fullBranchName)
            .maxRecords(maxEntries)
            .get();
    }

    public DiffResponse diff(String worldId,
            String fromBranch, String toBranch) {
        return nessieApi.getDiff()
            .fromRefName(formatBranchName(worldId, fromBranch))
            .toRefName(formatBranchName(worldId, toBranch))
            .get();
    }

    private String formatBranchName(
            String worldId, String branchName) {
        return worldId + "/" + branchName;
    }
}

分支命名策略：{worldId}/{branchName}。这利用了 Nessie 的命名空间隔离——不同 World 的同名分支（如都叫 "main"）在 Nessie 中是不同的引用。

异常转换：Nessie 的原生异常（NessieConflictException、NessieNotFoundException）被转换为 coomia-dip 的领域异常，由 gRPC 层统一映射。

#5. BranchManager：分支的 CRUD 与 Commit 日志

Java

@Component
@RequiredArgsConstructor
public class BranchManager {

    private final NessieClientWrapper nessieClient;
    private final WorldService worldService;

    public Branch createBranch(String worldId,
            String branchName, String sourceBranch) {
        // 验证 World 存在且 ACTIVE
        World world = worldService.getActive(worldId);

        // 验证源分支存在
        Branch source = nessieClient.getBranch(
            worldId, sourceBranch);

        // 从源分支的当前 hash 创建新分支
        return nessieClient.createBranch(
            worldId, branchName, source.getHash());
    }

    public List<CommitEntry> getCommitLog(String worldId,
            String branchName, int limit) {
        LogResponse log = nessieClient.getCommitLog(
            worldId, branchName, limit);

        return log.getLogEntries().stream()
            .map(entry -> CommitEntry.builder()
                .hash(entry.getCommitMeta().getHash())
                .message(entry.getCommitMeta().getMessage())
                .author(entry.getCommitMeta().getAuthor())
                .timestamp(entry.getCommitMeta()
                    .getCommitTime())
                .operations(entry.getOperations().stream()
                    .map(this::toOperation)
                    .collect(Collectors.toList()))
                .build())
            .collect(Collectors.toList());
    }

    public void deleteBranch(String worldId,
            String branchName) {
        if ("main".equals(branchName)) {
            throw new IllegalArgumentException(
                "Cannot delete the main branch");
        }
        nessieClient.deleteBranch(worldId, branchName);
    }
}

主分支保护：main 分支不可删除。这类似于 Git 仓库中保护 main/master 的做法——主分支是生产数据的唯一来源。

#6. MergeService：三路合并与冲突检测

Java

@Component
@RequiredArgsConstructor
public class MergeService {

    private final NessieClientWrapper nessieClient;

    public MergeResult mergeBranch(String worldId,
            String sourceBranch, String targetBranch,
            MergeStrategy strategy) {

        // 1. 获取 Diff
        DiffResponse diff = nessieClient.diff(
            worldId, sourceBranch, targetBranch);

        // 2. 检测冲突
        List<ConflictEntry> conflicts = detectConflicts(diff);

        if (!conflicts.isEmpty()) {
            if (strategy == MergeStrategy.FAIL_ON_CONFLICT) {
                return MergeResult.conflict(conflicts);
            }

            if (strategy == MergeStrategy.SOURCE_WINS) {
                // 源分支的更改覆盖目标分支
                resolveConflicts(conflicts,
                    ConflictResolution.USE_SOURCE);
            } else if (strategy == MergeStrategy.TARGET_WINS) {
                resolveConflicts(conflicts,
                    ConflictResolution.USE_TARGET);
            }
        }

        // 3. 执行合并
        try {
            nessieClient.merge(
                worldId, sourceBranch, targetBranch);
            return MergeResult.success(
                diff.getDiffs().size());
        } catch (NessieConflictException e) {
            return MergeResult.conflict(
                parseNessieConflicts(e));
        }
    }

    private List<ConflictEntry> detectConflicts(
            DiffResponse diff) {
        List<ConflictEntry> conflicts = new ArrayList<>();

        Map<ContentKey, DiffEntry> byKey = diff.getDiffs()
            .stream()
            .collect(Collectors.groupingBy(
                DiffEntry::getKey));

        for (Map.Entry<ContentKey, List<DiffEntry>> entry :
                byKey.entrySet()) {
            if (entry.getValue().size() > 1) {
                // 同一个 key 被两个分支同时修改
                conflicts.add(new ConflictEntry(
                    entry.getKey().toString(),
                    entry.getValue().get(0),
                    entry.getValue().get(1)));
            }
        }

        return conflicts;
    }
}

三种合并策略：

策略	行为
`FAIL_ON_CONFLICT`	有冲突则中止（安全策略）
`SOURCE_WINS`	源分支的变更覆盖目标
`TARGET_WINS`	目标分支的变更保留

这对标了 Git 的合并策略：FAIL_ON_CONFLICT 类似于默认的 Git merge（冲突时停止），SOURCE_WINS/TARGET_WINS 类似于 git merge -X theirs/ours。

#7. Release 管理：Tag 与快照

Java

@Override
public void createRelease(CreateReleaseRequest request,
        StreamObserver<ReleaseResponse> responseObserver) {
    try {
        String worldId = WorldContextHolder.getWorldId();

        // 获取源分支当前状态
        Branch source = nessieClient.getBranch(
            worldId, request.getSourceBranch());

        // 创建 Tag（不可变快照）
        Tag tag = nessieClient.createTag(
            worldId,
            request.getReleaseName(),   // e.g., "v2025.03.01"
            source.getHash()
        );

        // 记录 Release 元数据
        Release release = Release.builder()
            .releaseId(UUID.randomUUID().toString())
            .worldId(worldId)
            .releaseName(request.getReleaseName())
            .sourceBranch(request.getSourceBranch())
            .hash(tag.getHash())
            .description(request.getDescription())
            .createdBy(WorldContextHolder.getUserId())
            .build();

        worldService.saveRelease(release);

        responseObserver.onNext(
            worldMapper.toReleaseResponse(release));
        responseObserver.onCompleted();
    } catch (Exception e) {
        handleException(e, responseObserver);
    }
}

Release = Nessie Tag。Tag 是不可变的——它指向一个固定的 commit hash，永远不会改变。这为"某个时间点的数据全量快照"提供了保证，适用于审计、合规和回滚场景。

#8. 时间旅行查询

Java

@Override
public void queryAtTimestamp(QueryAtTimestampRequest request,
        StreamObserver<QueryResponse> responseObserver) {
    try {
        String worldId = WorldContextHolder.getWorldId();

        // 找到指定时间戳最近的 commit
        String hash = nessieClient.resolveHashAtTimestamp(
            worldId,
            request.getBranch(),
            Instant.ofEpochMilli(request.getTimestampMs())
        );

        // 构造指向历史版本的 WorldContext
        WorldContext historicalCtx = WorldContext.of(
            worldId,
            request.getBranch(),
            hash              // 指定 hash 而非 HEAD
        );

        // 执行查询
        QueryResult result = queryService.execute(
            request.getQuery(), historicalCtx);

        responseObserver.onNext(
            queryMapper.toResponse(result));
        responseObserver.onCompleted();
    } catch (Exception e) {
        handleException(e, responseObserver);
    }
}

时间旅行的核心是 WorldContext 可以携带一个特定的 hash。当查询服务收到带 hash 的 WorldContext 时，会从 Nessie 获取那个 hash 时刻的元数据快照，然后基于该快照执行查询。

#9. CounterfactualBranchService：假设分析

Java

@Component
public class CounterfactualBranchService {

    private final BranchManager branchManager;
    private final NessieClientWrapper nessieClient;

    public CounterfactualBranch createWhatIfBranch(
            String worldId, String name,
            String sourceBranch, String description) {

        // 创建隔离分支
        String branchName = "whatif-" + name;
        Branch branch = branchManager.createBranch(
            worldId, branchName, sourceBranch);

        return CounterfactualBranch.builder()
            .branchName(branchName)
            .worldId(worldId)
            .sourceBranch(sourceBranch)
            .sourceHash(branch.getHash())
            .description(description)
            .status(CounterfactualStatus.ACTIVE)
            .build();
    }

    public ComparisonResult compareWithSource(
            String worldId, String whatIfBranch) {
        // 获取 whatif 分支的源分支
        CounterfactualBranch cf = getCounterfactualBranch(
            worldId, whatIfBranch);

        // Diff：whatif vs source
        DiffResponse diff = nessieClient.diff(
            worldId, whatIfBranch, cf.getSourceBranch());

        // 分析变更影响
        return ComparisonResult.builder()
            .addedEntities(countAdded(diff))
            .modifiedEntities(countModified(diff))
            .deletedEntities(countDeleted(diff))
            .diff(diff)
            .build();
    }
}

反事实分析（What-If Analysis）是 coomia-dip 对标 Palantir Foundry Scenario 功能的实现。用户可以创建一个 whatif-* 分支，在其中自由修改数据，然后与源分支对比差异。

典型用例：

"如果我把预算增加 10%，销售额会变化多少？"
"如果裁掉这个部门，项目进度会受到什么影响？"

#10. 配额管理与资源限制

Java

@Override
public void getWorldQuota(GetWorldQuotaRequest request,
        StreamObserver<WorldQuotaResponse> responseObserver) {
    try {
        String worldId = WorldContextHolder.getWorldId();
        World world = worldService.getById(worldId);

        WorldQuota quota = worldService.getQuota(worldId);

        responseObserver.onNext(
            WorldQuotaResponse.newBuilder()
                .setMaxBranches(quota.getMaxBranches())
                .setCurrentBranches(
                    nessieClient.listBranches(worldId).size())
                .setMaxReleases(quota.getMaxReleases())
                .setCurrentReleases(
                    nessieClient.listTags(worldId).size())
                .setMaxEntities(quota.getMaxEntities())
                .setCurrentEntities(
                    worldService.countEntities(worldId))
                .setMaxStorageBytes(quota.getMaxStorageBytes())
                .setCurrentStorageBytes(
                    worldService.getStorageUsage(worldId))
                .build());
        responseObserver.onCompleted();
    } catch (Exception e) {
        handleException(e, responseObserver);
    }
}

每个 World 有四个维度的配额限制：分支数、Release 数、实体数、存储空间。超过配额时，World 会被自动挂起（SUSPENDED），直到配额调整或资源释放。

#11. Key Takeaways

World = Nessie Repository 隔离空间：每个 World 对应 Nessie 中的一组分支，数据完全隔离
分支命名约定 {worldId}/{branchName}：利用 Nessie 命名空间实现 World 间隔离
两阶段创建：先写数据库（CREATING），再创建 Nessie 分支，失败时有清理机制
三种合并策略：FAIL_ON_CONFLICT / SOURCE_WINS / TARGET_WINS，对标 Git 合并行为
Release = 不可变 Tag：固定 hash，适用于审计和合规
时间旅行 = WorldContext 携带 hash：查询服务透明支持历史版本查询
反事实分析分支：whatif-* 前缀命名，支持 Palantir Scenario 式的假设推演
四维配额管理：分支/Release/实体/存储，超配额自动挂起

#下一篇

S9-04：PolicyEngineService — 三模型权限的统一评估。我们将深入 Control Layer 的权限引擎，看它如何统一 ReBAC、ABAC、PBAC 三种权限模型的评估逻辑。

Tags: #coomia-dip #source-code-reading #control-Layer #nessie #git-semantics #branching #merge #time-travel #what-if-analysis #world-context