Source Code Reading: WorldManagerService — Git Operations for Data Worlds

“

Series: S9 Source Code Reading · Article 3 | Level: Advanced | Reading Time: 25 min

#TL;DR

WorldManagerService is the core service in the Control Layer managing "data worlds" (Worlds) through Nessie integration, implementing Git-like data version management. Each World is a fully isolated data space supporting branching, merging, releases (Tags), time travel, and counterfactual analysis. This article analyzes the Nessie integration architecture, World lifecycle management, BranchManager operations, MergeService conflict detection with three-way merge strategies, CounterfactualBranchService for hypothesis analysis, and quota management mechanisms.

#Table of Contents

World Concept and Architecture Position
Service Structure: Six Collaborators
World Lifecycle Management
Nessie Integration: NessieClientWrapper
BranchManager: Branch CRUD and Commit Log
MergeService: Three-Way Merge and Conflict Detection
Release Management: Tags and Snapshots
Time Travel Queries
CounterfactualBranchService: Hypothesis Analysis
Quota Management and Resource Limits
Key Takeaways

#1. World Concept and Architecture Position

In coomia-dip, World is the top-level data isolation unit. It corresponds to Palantir Foundry's Namespace/Project concept, but with added Git semantics:

Code

World "production"
├── Branch "main"           # Main branch (production data)
├── Branch "dev-feature-x"  # Development branch
├── Branch "whatif-budget"   # Counterfactual analysis branch
├── Tag "v2025.03.01"       # Release snapshot
└── Tag "v2025.02.15"       # Historical release

Each World corresponds to an isolated space within a Nessie Repository. Nessie is coomia-dip's "data catalog version control system" — it does not store data itself (data lives in Doris/Iceberg), but manages metadata versions (which tables exist, what their schemas are, version snapshot IDs for each record).

#2. Service Structure: Six Collaborators

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;
}

Collaborator	Responsibility
`WorldService`	World CRUD, state management
`BranchManager`	Branch creation/deletion/listing/commit log
`MergeService`	Branch merging, conflict detection
`NessieClientWrapper`	Nessie REST API wrapper
`CounterfactualBranchService`	Counterfactual (What-If) analysis branches
`WorldMapper`	Proto <-> Domain mapping

Design intent: Complex World management is split into multiple single-responsibility services rather than one massive God Class. WorldManagerServiceImpl handles only gRPC protocol adaptation and request routing.

#3. World Lifecycle Management

Java

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

        // 1. Quota check
        worldService.checkQuota(WorldContextHolder.getTenantId());

        // 2. Create World record
        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. Create corresponding branch in Nessie
        nessieClient.createBranch(
            created.getWorldId(),
            "main",          // Default main branch
            null             // Start from empty state
        );

        // 4. Update status to ACTIVE
        created.setStatus(WorldStatus.ACTIVE);
        worldService.update(created);

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

Two-phase creation: First creates the World record (status CREATING), then creates the Nessie branch. If the Nessie call fails, the World record remains in CREATING status, and a background task periodically cleans up these "zombie" Worlds.

Java

public enum WorldStatus {
    CREATING,     // Nessie branch being created
    ACTIVE,       // Normal usage
    SUSPENDED,    // Suspended (over quota, etc.)
    DELETING,     // Being deleted
    DELETED       // Deleted
}

#4. Nessie Integration: 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;
    }
}

Branch naming strategy: {worldId}/{branchName}. This leverages Nessie's namespace isolation — same-named branches in different Worlds (e.g., both named "main") are different references in Nessie.

Exception translation: Nessie's native exceptions (NessieConflictException, NessieNotFoundException) are converted to coomia-dip domain exceptions, unified by the gRPC layer.

#5. BranchManager: Branch CRUD and Commit Log

Java

@Component
@RequiredArgsConstructor
public class BranchManager {

    private final NessieClientWrapper nessieClient;
    private final WorldService worldService;

    public Branch createBranch(String worldId,
            String branchName, String sourceBranch) {
        World world = worldService.getActive(worldId);
        Branch source = nessieClient.getBranch(worldId, sourceBranch);
        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 branch protection: The main branch cannot be deleted. Similar to protecting main/master in Git repositories — the main branch is the single source of truth for production data.

#6. MergeService: Three-Way Merge and Conflict Detection

Java

@Component
@RequiredArgsConstructor
public class MergeService {

    private final NessieClientWrapper nessieClient;

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

        DiffResponse diff = nessieClient.diff(
            worldId, sourceBranch, targetBranch);

        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);
            }
        }

        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, List<DiffEntry>> byKey = diff.getDiffs()
            .stream()
            .collect(Collectors.groupingBy(DiffEntry::getKey));

        for (Map.Entry<ContentKey, List<DiffEntry>> entry : byKey.entrySet()) {
            if (entry.getValue().size() > 1) {
                conflicts.add(new ConflictEntry(
                    entry.getKey().toString(),
                    entry.getValue().get(0),
                    entry.getValue().get(1)));
            }
        }
        return conflicts;
    }
}

Three merge strategies:

Strategy	Behavior
`FAIL_ON_CONFLICT`	Abort on conflict (safe strategy)
`SOURCE_WINS`	Source branch changes override target
`TARGET_WINS`	Target branch changes are preserved

This mirrors Git merge strategies: FAIL_ON_CONFLICT is like default Git merge (stops on conflict), SOURCE_WINS/TARGET_WINS are like git merge -X theirs/ours.

#7. Release Management: Tags and Snapshots

Java

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

        Tag tag = nessieClient.createTag(
            worldId,
            request.getReleaseName(),   // e.g., "v2025.03.01"
            source.getHash()
        );

        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. Tags are immutable — they point to a fixed commit hash that never changes. This provides a guarantee of "full data snapshot at a point in time," suitable for audit, compliance, and rollback scenarios.

#8. Time Travel Queries

Java

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

        String hash = nessieClient.resolveHashAtTimestamp(
            worldId,
            request.getBranch(),
            Instant.ofEpochMilli(request.getTimestampMs())
        );

        WorldContext historicalCtx = WorldContext.of(
            worldId,
            request.getBranch(),
            hash              // Specific hash, not HEAD
        );

        QueryResult result = queryService.execute(
            request.getQuery(), historicalCtx);

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

The core of time travel is that WorldContext can carry a specific hash. When the query service receives a WorldContext with a hash, it retrieves the metadata snapshot from Nessie at that hash point and executes the query against that snapshot.

#9. CounterfactualBranchService: Hypothesis Analysis

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) {
        CounterfactualBranch cf = getCounterfactualBranch(
            worldId, whatIfBranch);

        DiffResponse diff = nessieClient.diff(
            worldId, whatIfBranch, cf.getSourceBranch());

        return ComparisonResult.builder()
            .addedEntities(countAdded(diff))
            .modifiedEntities(countModified(diff))
            .deletedEntities(countDeleted(diff))
            .diff(diff)
            .build();
    }
}

Counterfactual analysis (What-If Analysis) is coomia-dip's implementation of Palantir Foundry's Scenario feature. Users create a whatif-* branch, freely modify data within it, then compare differences with the source branch.

Typical use cases:

"If I increase the budget by 10%, how will sales change?"
"If I eliminate this department, what impact will it have on project timelines?"

#10. Quota Management and Resource Limits

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);
    }
}

Each World has four dimensions of quota limits: branches, releases, entities, and storage. When quotas are exceeded, the World is automatically suspended (SUSPENDED) until quotas are adjusted or resources freed.

#11. Key Takeaways

World = Nessie Repository isolation space: Each World maps to a set of branches in Nessie with complete data isolation
Branch naming convention {worldId}/{branchName}: Leverages Nessie namespaces for World-level isolation
Two-phase creation: Database first (CREATING), then Nessie branch, with cleanup mechanism on failure
Three merge strategies: FAIL_ON_CONFLICT / SOURCE_WINS / TARGET_WINS, mirroring Git merge behavior
Release = immutable Tag: Fixed hash, suitable for audit and compliance
Time travel = WorldContext with hash: Query services transparently support historical version queries
Counterfactual analysis branches: whatif-* prefix naming, supporting Palantir Scenario-style hypothesis exploration
Four-dimensional quota management: Branches/Releases/Entities/Storage, auto-suspension on quota breach

#Next Article

S9-04: PolicyEngineService — Unified Three-Model Permission Evaluation. We will dive into the Control Layer's permission engine to see how it unifies ReBAC, ABAC, and PBAC evaluation logic.

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