Traditional Data Management:

Timeline ──────────────────────────────►

  T1: Data correct    T2: Someone broke it  T3: Problem found
  ┌──────┐            ┌──────┐             ┌──────┐
  │ OK   │   ──►      │ BUG! │   ──►       │ ???  │
  │      │  modify     │      │  discover    │ Can't│
  │      │            │      │              │ roll │
  └──────┘            └──────┘              │ back │
                                            └──────┘
Problems:
- Who changed it?   -> Unknown
- What changed?     -> Unknown
- Can we roll back? -> No
- Parallel edits?   -> No

Nessie + Iceberg Architecture:

┌─────────────────────────────────────────┐
│              Nessie Server              │
│  ┌─────────────────────────────────┐    │
│  │  Git-Like Catalog API           │    │
│  │  - Branch management            │    │
│  │  - Commit history               │    │
│  │  - Merge operations             │    │
│  │  - Tag management               │    │
│  └──────────────┬──────────────────┘    │
│                 │                        │
│  ┌──────────────┴──────────────────┐    │
│  │  Version Store (RocksDB/JDBC)   │    │
│  │  - Commit graph                 │    │
│  │  - Branch pointers              │    │
│  │  - Table metadata refs          │    │
│  └─────────────────────────────────┘    │
└────────────────────┬────────────────────┘
                     │ points to
                     v
┌─────────────────────────────────────────┐
│           Apache Iceberg                │
│  ┌─────────────────────────────────┐    │
│  │  Table Metadata                 │    │
│  │  - Schema evolution             │    │
│  │  - Partition spec               │    │
│  │  - Snapshot history             │    │
│  └──────────────┬──────────────────┘    │
│                 │                        │
│  ┌──────────────┴──────────────────┐    │
│  │  Data Files (Parquet on MinIO)  │    │
│  │  - Columnar storage             │    │
│  │  - Statistics per file           │    │
│  │  - Delete files (MoR)           │    │
│  └─────────────────────────────────┘    │
└─────────────────────────────────────────┘

World-to-Branch Mapping:

Nessie Branches:

  main ──●──●──●──●──●──●──●──●──●──►    (production world)
              │           ▲
              │           │ merge
              └──●──●──●──┘               (what-if analysis)
                 │
                 world-whatif-001

  world-dev ──●──●──●──●──►               (development world)
                    │
                    └──●──●──►            (feature branch)
                       world-feature-x

# Branch management in WorldManagerService
class WorldManagerService:
    """Manages World-to-Nessie-Branch mapping"""

    def __init__(self, nessie_client: NessieClient):
        self.nessie = nessie_client

    async def create_world(
        self,
        world_id: str,
        display_name: str,
        source_world_id: str | None = None
    ) -> World:
        """Create a new World (Nessie branch)"""
        branch_name = f"world-{world_id}"

        if source_world_id:
            # Derive from existing world (like git checkout -b)
            source_branch = f"world-{source_world_id}"
            source_ref = await self.nessie.get_reference(source_branch)
            await self.nessie.create_reference(
                branch_name=branch_name,
                source_hash=source_ref.hash
            )
        else:
            # Create from main branch
            main_ref = await self.nessie.get_reference("main")
            await self.nessie.create_reference(
                branch_name=branch_name,
                source_hash=main_ref.hash
            )

        return World(
            id=world_id,
            display_name=display_name,
            branch_name=branch_name,
            created_at=datetime.utcnow()
        )

Commit History Example:

world-prod-001 branch:

  commit: abc123  "Add 500 equipment records"
    │
    ├── entity_common: +500 rows
    └── entity_edge: +1200 rows (equipment-department relations)

  commit: def456  "Update equipment status to repaired"
    │
    └── entity_common: ~50 rows (status: broken -> repaired)

  commit: ghi789  "Delete decommissioned equipment"
    │
    ├── entity_common: -30 rows
    └── entity_edge: -75 rows

async def commit_changes(
    self,
    world_id: str,
    operations: list[DataOperation],
    message: str,
    author: str
) -> CommitResult:
    """Commit data changes to the World's Nessie branch"""
    branch_name = f"world-{world_id}"
    branch_ref = await self.nessie.get_reference(branch_name)

    # Build Iceberg operations
    iceberg_ops = []
    for op in operations:
        if op.type == OperationType.INSERT:
            iceberg_ops.append(
                IcebergAppend(table=op.table, data_files=op.files)
            )
        elif op.type == OperationType.UPDATE:
            iceberg_ops.append(
                IcebergOverwrite(
                    table=op.table,
                    delete_files=op.old_files,
                    data_files=op.new_files
                )
            )
        elif op.type == OperationType.DELETE:
            iceberg_ops.append(
                IcebergDelete(table=op.table, delete_files=op.files)
            )

    # Commit to Nessie
    result = await self.nessie.commit(
        branch=branch_name,
        expected_hash=branch_ref.hash,
        operations=iceberg_ops,
        commit_meta=CommitMeta(
            message=message,
            author=author,
            timestamp=datetime.utcnow()
        )
    )

    return CommitResult(
        commit_hash=result.hash,
        branch=branch_name,
        message=message,
        timestamp=result.timestamp
    )

Release-to-Tag Mapping:

  main ──●──●──●──●──●──●──●──●──●──►
              │        │        │
              v        v        v
           tag:      tag:      tag:
           v1.0      v1.1      v2.0
           (Q1 rpt)  (fix)     (annual)

async def create_release(
    self,
    world_id: str,
    release_name: str,
    description: str
) -> Release:
    """Create a Release (Nessie Tag)"""
    branch_name = f"world-{world_id}"
    branch_ref = await self.nessie.get_reference(branch_name)

    tag_name = f"release-{world_id}-{release_name}"

    await self.nessie.create_tag(
        tag_name=tag_name,
        hash=branch_ref.hash
    )

    return Release(
        name=release_name,
        tag=tag_name,
        commit_hash=branch_ref.hash,
        description=description,
        created_at=datetime.utcnow()
    )

Three-Way Merge Diagram:

           Common Ancestor (Base)
              │
         ┌────┴────┐
         │         │
    main branch  what-if branch
    modified A   modified A (CONFLICT!)
    modified B   modified C (no conflict)
         │         │
         └────┬────┘
              │
         Merge Result
    A: needs conflict resolution
    B: take main's modification
    C: take what-if's modification

class ThreeWayMerger:
    """Three-way merge implementation"""

    async def merge(
        self,
        source_branch: str,
        target_branch: str,
        strategy: MergeStrategy = MergeStrategy.NORMAL
    ) -> MergeResult:
        """Execute three-way merge"""
        # Get three-way references
        source_ref = await self.nessie.get_reference(source_branch)
        target_ref = await self.nessie.get_reference(target_branch)
        base_ref = await self.nessie.find_common_ancestor(
            source_ref.hash, target_ref.hash
        )

        # Compute diffs
        source_diff = await self.compute_diff(base_ref.hash, source_ref.hash)
        target_diff = await self.compute_diff(base_ref.hash, target_ref.hash)

        # Detect conflicts
        conflicts = self.detect_conflicts(source_diff, target_diff)

        if conflicts and strategy == MergeStrategy.NORMAL:
            return MergeResult(
                status=MergeStatus.CONFLICT,
                conflicts=conflicts,
                message="Merge conflicts detected"
            )

        # No conflicts or auto-resolution strategy
        if strategy == MergeStrategy.THEIRS:
            resolved = self.resolve_with_theirs(conflicts)
        elif strategy == MergeStrategy.OURS:
            resolved = self.resolve_with_ours(conflicts)
        else:
            resolved = []

        # Execute merge
        result = await self.nessie.merge(
            from_branch=source_branch,
            to_branch=target_branch,
            expected_hash=target_ref.hash
        )

        return MergeResult(
            status=MergeStatus.SUCCESS,
            commit_hash=result.hash,
            merged_changes=len(source_diff) + len(target_diff),
            resolved_conflicts=len(resolved)
        )

    def detect_conflicts(
        self,
        source_diff: list[DiffEntry],
        target_diff: list[DiffEntry]
    ) -> list[Conflict]:
        """Detect conflicts: same entity modified in both branches"""
        conflicts = []
        source_keys = {d.entity_key for d in source_diff}
        target_keys = {d.entity_key for d in target_diff}

        overlapping = source_keys & target_keys

        for key in overlapping:
            source_change = next(d for d in source_diff if d.entity_key == key)
            target_change = next(d for d in target_diff if d.entity_key == key)

            # Only a conflict if the same fields were modified
            if self.has_field_conflict(source_change, target_change):
                conflicts.append(Conflict(
                    entity_key=key,
                    source_change=source_change,
                    target_change=target_change
                ))

        return conflicts

class FieldLevelMerger:
    """Fine-grained field-level merging"""

    def merge_entity(
        self,
        base: dict,
        source: dict,
        target: dict
    ) -> tuple[dict, list[str]]:
        """
        Three-way field-level merge

        base:   common ancestor version
        source: source branch version
        target: target branch version

        Returns: (merged result, list of conflicting fields)
        """
        merged = dict(base)
        conflicts = []

        all_fields = set(base.keys()) | set(source.keys()) | set(target.keys())

        for field in all_fields:
            base_val = base.get(field)
            source_val = source.get(field)
            target_val = target.get(field)

            if source_val == target_val:
                # Both sides made the same change
                merged[field] = source_val
            elif source_val == base_val:
                # Only target modified, take target
                merged[field] = target_val
            elif target_val == base_val:
                # Only source modified, take source
                merged[field] = source_val
            else:
                # Real conflict: both changed differently
                conflicts.append(field)
                merged[field] = target_val  # default to target

        return merged, conflicts

-- View data at a specific commit
SELECT * FROM entity_common
AT BRANCH 'world-prod-001'
AT COMMIT 'abc123def456'
WHERE object_type_id = 'Equipment'
LIMIT 100;

-- View data as of January 15, 2026
SELECT * FROM entity_common
AT BRANCH 'world-prod-001'
AS OF TIMESTAMP '2026-01-15T10:00:00Z'
WHERE object_type_id = 'Equipment';

class TemporalQueryService:
    """Time-travel query service"""

    async def query_at_commit(
        self,
        world_id: str,
        commit_hash: str,
        query: str
    ) -> QueryResult:
        """Query data at a specific commit point"""
        branch_name = f"world-{world_id}"

        # Get Iceberg snapshot for the commit
        ref = await self.nessie.get_reference(
            branch_name,
            hash_on_ref=commit_hash
        )

        # Get Iceberg table metadata at that point
        table_metadata = await self.nessie.get_table_metadata(
            ref=ref,
            table_name="entity_common"
        )

        # Execute query with the corresponding snapshot
        snapshot_id = table_metadata.current_snapshot_id
        return await self.iceberg_engine.query(
            table="entity_common",
            snapshot_id=snapshot_id,
            sql=query
        )

    async def query_at_timestamp(
        self,
        world_id: str,
        timestamp: datetime,
        query: str
    ) -> QueryResult:
        """Query data at a specific timestamp"""
        branch_name = f"world-{world_id}"

        # Find the commit closest to the timestamp
        log = await self.nessie.get_commit_log(
            branch_name,
            filter=f"timestamp <= {timestamp.isoformat()}"
        )

        if not log.entries:
            raise ValueError(f"No commits found before {timestamp}")

        commit_hash = log.entries[0].hash
        return await self.query_at_commit(world_id, commit_hash, query)

Nessie REST API Endpoints:

┌─────────────────────────────────────────────────┐
│ Trees API (Branch & Tag Management)             │
├─────────────────────────────────────────────────┤
│ GET    /api/v2/trees                            │
│ POST   /api/v2/trees                            │
│ GET    /api/v2/trees/{name}                     │
│ DELETE /api/v2/trees/{name}                     │
│ PUT    /api/v2/trees/{name}                     │
├─────────────────────────────────────────────────┤
│ Content API (Table Content Management)          │
├─────────────────────────────────────────────────┤
│ GET    /api/v2/trees/{name}/contents            │
│ GET    /api/v2/trees/{name}/contents/{key}      │
│ POST   /api/v2/trees/{name}/contents            │
├─────────────────────────────────────────────────┤
│ Commit API (Commit Management)                  │
├─────────────────────────────────────────────────┤
│ GET    /api/v2/trees/{name}/history             │
│ POST   /api/v2/trees/{name}/history/transplant  │
│ POST   /api/v2/trees/{name}/history/merge       │
├─────────────────────────────────────────────────┤
│ Diff API (Change Comparison)                    │
├─────────────────────────────────────────────────┤
│ GET    /api/v2/trees/{from}/diff/{to}           │
└─────────────────────────────────────────────────┘

class NessieClient:
    """Nessie API client wrapper"""

    def __init__(self, base_url: str, auth_token: str | None = None):
        self.base_url = base_url.rstrip("/")
        self.session = aiohttp.ClientSession(
            headers={"Authorization": f"Bearer {auth_token}"} if auth_token else {}
        )

    # === Branch Operations ===

    async def list_branches(self) -> list[Branch]:
        resp = await self.session.get(
            f"{self.base_url}/api/v2/trees",
            params={"type": "BRANCH"}
        )
        data = await resp.json()
        return [Branch(**ref) for ref in data["references"]]

    async def create_branch(self, name: str, source_hash: str) -> Branch:
        resp = await self.session.post(
            f"{self.base_url}/api/v2/trees",
            json={"type": "BRANCH", "name": name, "hash": source_hash}
        )
        return Branch(**(await resp.json()))

    async def delete_branch(self, name: str, hash: str) -> None:
        await self.session.delete(
            f"{self.base_url}/api/v2/trees/{name}",
            params={"type": "BRANCH", "expectedHash": hash}
        )

    # === Commit Operations ===

    async def commit(
        self,
        branch: str,
        expected_hash: str,
        operations: list[Operation],
        commit_meta: CommitMeta
    ) -> CommitResponse:
        resp = await self.session.post(
            f"{self.base_url}/api/v2/trees/{branch}/history/commit",
            json={
                "expectedHash": expected_hash,
                "operations": [op.to_dict() for op in operations],
                "commitMeta": commit_meta.to_dict()
            }
        )
        return CommitResponse(**(await resp.json()))

    async def get_commit_log(
        self, branch: str, max_records: int = 100, filter: str | None = None
    ) -> CommitLog:
        params = {"maxRecords": max_records}
        if filter:
            params["filter"] = filter
        resp = await self.session.get(
            f"{self.base_url}/api/v2/trees/{branch}/history",
            params=params
        )
        return CommitLog(**(await resp.json()))

    # === Merge Operations ===

    async def merge(
        self, from_branch: str, to_branch: str, expected_hash: str
    ) -> MergeResponse:
        from_ref = await self.get_reference(from_branch)
        resp = await self.session.post(
            f"{self.base_url}/api/v2/trees/{to_branch}/history/merge",
            json={
                "fromRefName": from_branch,
                "fromHash": from_ref.hash,
                "expectedHash": expected_hash
            }
        )
        return MergeResponse(**(await resp.json()))

    # === Diff Operations ===

    async def diff(self, from_ref: str, to_ref: str) -> DiffResponse:
        resp = await self.session.get(
            f"{self.base_url}/api/v2/trees/{from_ref}/diff/{to_ref}"
        )
        return DiffResponse(**(await resp.json()))

Iceberg Snapshot Chain:

  S1 ──► S2 ──► S3 ──► S4 (current)
  │       │      │      │
  │       │      │      └── manifest-list-4
  │       │      │           ├── manifest-a: [file-7, file-8]
  │       │      │           └── manifest-b: [file-5, file-6]
  │       │      │
  │       │      └── manifest-list-3
  │       │           └── manifest-a: [file-5, file-6]
  │       │
  │       └── manifest-list-2
  │            └── manifest-a: [file-3, file-4]
  │
  └── manifest-list-1
       └── manifest-a: [file-1, file-2]

File Layout on MinIO:
  s3://lakehouse/
  ├── entity_common/
  │   ├── metadata/
  │   │   ├── v1.metadata.json
  │   │   ├── v2.metadata.json
  │   │   ├── v3.metadata.json
  │   │   └── v4.metadata.json (current)
  │   ├── data/
  │   │   ├── file-1.parquet
  │   │   ├── ...
  │   │   └── file-8.parquet
  │   └── manifests/
  │       ├── manifest-list-1.avro
  │       ├── ...
  │       └── manifest-b.avro
  └── entity_edge/
      └── ...

class IcebergSchemaManager:
    """Iceberg schema evolution management"""

    async def add_column(
        self, table_name: str, column_name: str,
        column_type: str, comment: str | None = None
    ) -> None:
        """Add new column (no impact on existing data)"""
        table = self.catalog.load_table(table_name)
        with table.update_schema() as update:
            update.add_column(column_name, column_type, comment)

    async def rename_column(
        self, table_name: str, old_name: str, new_name: str
    ) -> None:
        """Rename column (leveraging Iceberg's ID mapping)"""
        table = self.catalog.load_table(table_name)
        with table.update_schema() as update:
            update.rename_column(old_name, new_name)

    async def evolve_partition(
        self, table_name: str, new_partition_spec: list[PartitionField]
    ) -> None:
        """Partition evolution (no data rewrite needed)"""
        table = self.catalog.load_table(table_name)
        with table.update_spec() as update:
            for field in new_partition_spec:
                update.add_field(field.source, field.transform, field.name)

WorldManagerService Architecture:

┌─────────────────────────────────────────┐
│         WorldManagerService             │
│                                         │
│  ┌──────────┐  ┌────────────────────┐   │
│  │ World    │  │ Branch             │   │
│  │ CRUD     │  │ Management         │   │
│  └────┬─────┘  └────────┬───────────┘   │
│       │                 │               │
│  ┌────┴─────┐  ┌────────┴───────────┐   │
│  │ Release  │  │ Merge              │   │
│  │ Mgmt     │  │ Service            │   │
│  └────┬─────┘  └────────┬───────────┘   │
│       │                 │               │
│  ┌────┴─────────────────┴───────────┐   │
│  │      NessieClient                │   │
│  └──────────────┬───────────────────┘   │
│                 │                       │
└─────────────────┼───────────────────────┘
                  │ HTTP/REST
                  v
         ┌────────────────┐
         │  Nessie Server │
         └────────────────┘

// world_manager.proto
service WorldManagerService {
    rpc CreateWorld(CreateWorldRequest) returns (WorldResponse);
    rpc GetWorld(GetWorldRequest) returns (WorldResponse);
    rpc ListWorlds(ListWorldsRequest) returns (ListWorldsResponse);
    rpc DeleteWorld(DeleteWorldRequest) returns (Empty);

    rpc ForkWorld(ForkWorldRequest) returns (WorldResponse);
    rpc MergeWorld(MergeWorldRequest) returns (MergeResponse);

    rpc CreateRelease(CreateReleaseRequest) returns (ReleaseResponse);
    rpc ListReleases(ListReleasesRequest) returns (ListReleasesResponse);

    rpc GetCommitHistory(CommitHistoryRequest) returns (CommitHistoryResponse);
    rpc GetDiff(DiffRequest) returns (DiffResponse);
}

message ForkWorldRequest {
    string source_world_id = 1;
    string new_world_id = 2;
    string display_name = 3;
    string description = 4;
}

message MergeWorldRequest {
    string source_world_id = 1;
    string target_world_id = 2;
    MergeStrategy strategy = 3;
    string message = 4;
}

enum MergeStrategy {
    NORMAL = 0;
    OURS = 1;
    THEIRS = 2;
    FIELD_LEVEL = 3;
}

What-If Analysis Workflow:

1. Create analysis branch
   main ──●──●──●──► (production data)
               │
               └──► world-whatif-001 (analysis branch)

2. Modify parameters on analysis branch
   world-whatif-001: Modify pricing model parameters
                     Run simulation
                     Generate predictions

3. Compare analysis results
   diff(main, world-whatif-001)
   -> Revenue change: +12%
   -> Customer churn risk: -5%

4. Decision: adopt or discard
   If adopt:  merge world-whatif-001 -> main
   If discard: delete world-whatif-001

async def audit_entity_changes(
    self,
    world_id: str,
    entity_id: str,
    start_time: datetime,
    end_time: datetime
) -> list[AuditEntry]:
    """Audit entity change history"""
    branch_name = f"world-{world_id}"

    log = await self.nessie.get_commit_log(
        branch_name,
        filter=f"timestamp >= {start_time.isoformat()} "
               f"AND timestamp <= {end_time.isoformat()}"
    )

    audit_entries = []
    for i, entry in enumerate(log.entries):
        if i + 1 < len(log.entries):
            diff = await self.nessie.diff(
                from_ref=f"{branch_name}@{log.entries[i+1].hash}",
                to_ref=f"{branch_name}@{entry.hash}"
            )
            entity_changes = [
                d for d in diff.diffs if d.key.contains(entity_id)
            ]
            if entity_changes:
                audit_entries.append(AuditEntry(
                    commit_hash=entry.hash,
                    timestamp=entry.timestamp,
                    author=entry.author,
                    message=entry.message,
                    changes=entity_changes
                ))

    return audit_entries

async def rollback_world(
    self, world_id: str, target_commit: str, reason: str
) -> CommitResult:
    """Roll back a World to a specific commit"""
    branch_name = f"world-{world_id}"
    current_ref = await self.nessie.get_reference(branch_name)

    # Create rollback branch
    rollback_branch = f"{branch_name}-rollback-{target_commit[:8]}"
    await self.nessie.create_branch(rollback_branch, target_commit)

    # Merge rollback branch back (THEIRS strategy)
    result = await self.nessie.merge(
        from_branch=rollback_branch,
        to_branch=branch_name,
        expected_hash=current_ref.hash
    )

    # Clean up rollback branch
    rollback_ref = await self.nessie.get_reference(rollback_branch)
    await self.nessie.delete_branch(rollback_branch, rollback_ref.hash)

    return CommitResult(
        commit_hash=result.hash,
        branch=branch_name,
        message=f"Rollback to {target_commit[:8]}: {reason}"
    )

class NessieGarbageCollector:
    """Nessie + Iceberg garbage collection"""

    async def collect(
        self, max_age_days: int = 30, dry_run: bool = True
    ) -> GCResult:
        """Clean up expired snapshots and data files"""
        active_refs = await self.get_all_active_references()

        active_snapshots = set()
        for ref in active_refs:
            snapshots = await self.get_reachable_snapshots(ref)
            active_snapshots.update(snapshots)

        all_snapshots = await self.get_all_snapshots()
        expired = [
            s for s in all_snapshots
            if s.id not in active_snapshots
            and s.timestamp < datetime.utcnow() - timedelta(days=max_age_days)
        ]

        if dry_run:
            return GCResult(
                expired_snapshots=len(expired),
                reclaimable_bytes=sum(s.size for s in expired),
                deleted=False
            )

        for snapshot in expired:
            await self.delete_snapshot_files(snapshot)

        return GCResult(
            expired_snapshots=len(expired),
            reclaimed_bytes=sum(s.size for s in expired),
            deleted=True
        )

Nessie + Iceberg Integration in coomia-dip:

┌─────────────┐    ┌──────────────┐
│ Control     │    │ Intelligence │
│ Layer (B)   │    │  Layer (D)   │
│             │    │              │
│ WorldMgr    │    │ What-If      │
│ Service     │    │ Analysis     │
└──────┬──────┘    └──────┬───────┘
       │                  │
       │    gRPC          │  gRPC
       │                  │
┌──────┴──────────────────┴───────┐
│        Data Layer (C)           │
│                                  │
│  ┌────────────────────────────┐  │
│  │  NessieIcebergService      │  │
│  │  - Branch CRUD             │  │
│  │  - Commit & Merge          │  │
│  │  - Time Travel             │  │
│  │  - Diff                    │  │
│  └──────────────┬─────────────┘  │
│                 │                 │
│  ┌──────────────┴─────────────┐  │
│  │     Nessie Server          │  │
│  │     + Iceberg Catalog      │  │
│  └──────────────┬─────────────┘  │
│                 │                 │
│  ┌──────────────┴─────────────┐  │
│  │     MinIO (S3 Storage)     │  │
│  └────────────────────────────┘  │
└──────────────────────────────────┘