World Transform vs Pipeline Transform:

Pipeline Transform (local):
  Input:  One data source (e.g., MySQL table)
  Output: One target (e.g., Ontology Entity Type)
  Scope:  Single data stream
  Txn:    Pipeline-internal consistency

World Transform (global):
  Input:  Entire Ontology world state (multiple Entity Types + graph)
  Output: New world state (atomic update)
  Scope:  Cross Entity Type, cross relationship global computation
  Txn:    Global consistency (all succeed or all rollback)

Example:
  "Recalculate risk ratings for all customers"
  -> Needs to read Customer, Transaction, RiskModel
  -> Update Customer.risk_level
  -> Update RiskAlert relationships
  -> Update RiskScore metrics
  -> All above must complete atomically

from onto_transform import WorldTransform, Input, Output

@WorldTransform(
    name="risk-scoring",
    description="Recalculate risk scores for all customers",
    schedule="0 2 * * *",  # daily at 2 AM
    version="2.0.0",
)
class RiskScoringTransform:
    customers = Input("Customer", properties=["id", "credit_rating", "region"])
    transactions = Input("Transaction", properties=["customer_id", "amount", "type"])
    risk_model = Input("RiskModel", properties=["id", "weights", "thresholds"])

    risk_scores = Output("Customer", properties=["risk_level", "risk_score"])
    risk_alerts = Output("RiskAlert", relationship=True)

    def compute(self, ctx: TransformContext):
        customers = ctx.read(self.customers)
        transactions = ctx.read(self.transactions)
        model = ctx.read(self.risk_model).first()

        for customer in customers:
            customer_txns = transactions.filter(customer_id=customer.id)
            risk_score = self._calculate_risk(customer, customer_txns, model)

            ctx.update(self.risk_scores, customer.id, {
                'risk_level': self._score_to_level(risk_score),
                'risk_score': risk_score,
            })

            if risk_score > model.thresholds['high']:
                ctx.create_relationship(self.risk_alerts, {
                    'source_id': customer.id,
                    'target_id': f"alert-{customer.id}-{ctx.run_id}",
                    'alert_type': 'HIGH_RISK',
                    'score': risk_score,
                })

World Transform Dependency Graph:

Transforms declare Input and Output Entity Types
System auto-builds dependency graph:

  risk-model-training
    Output: RiskModel
         |
         v
  risk-scoring (this Transform)
    Input:  Customer, Transaction, RiskModel
    Output: Customer.risk_level, RiskAlert
         |
         v
  risk-reporting
    Input:  Customer.risk_level, RiskAlert
    Output: RiskReport

Dependency graph ensures:
  1. risk-model-training runs before risk-scoring
  2. risk-scoring runs before risk-reporting
  3. Circular dependencies are detected and rejected

class WorldTransformExecutor:
    """World Transform transaction executor"""

    async def execute(self, transform: WorldTransform) -> TransformResult:
        branch_name = f"wt-{transform.name}-{uuid4().hex[:8]}"
        await self._nessie.create_branch(branch_name, from_ref="main")

        try:
            ctx = TransformContext(
                branch=branch_name,
                nessie=self._nessie,
                iceberg=self._iceberg,
            )
            transform.compute(ctx)

            validation = await self._validate_output(ctx)
            if not validation.passed:
                raise TransformValidationError(validation.errors)

            await self._nessie.merge(
                from_branch=branch_name,
                to_branch="main",
                conflict_resolution="REJECT",
            )

            return TransformResult(
                status="SUCCESS",
                branch=branch_name,
                changes=ctx.get_change_summary(),
            )

        except Exception as e:
            await self._nessie.delete_branch(branch_name)
            return TransformResult(status="FAILED", error=str(e))

World Transform Conflict Scenarios:

Two Transforms concurrently modify the same Customer:

  main --*------------------------------ HEAD
          \
           \-- wt-risk-scoring (modifies Customer.risk_level)
          \
           \-- wt-segmentation (modifies Customer.segment)

Merge strategies:
  1. First-completed merges to main
  2. Second detects conflict (Customer already modified)
  3. Conflict handling options:
     a) REJECT: fail, let scheduler retry
     b) RETRY: re-read latest data, re-execute
     c) PROPERTY_LEVEL: different properties don't conflict
        risk_level and segment are different -> auto-merge

class IncrementalWorldTransform:
    """Incremental World Transform — process only changed data"""

    @WorldTransform(name="risk-scoring-incremental", incremental=True)
    class IncrementalRiskScoring:
        customers = Input("Customer", incremental=True)
        transactions = Input("Transaction", incremental=True)

        def compute(self, ctx: TransformContext):
            changed_customers = ctx.read_changes(self.customers)
            new_transactions = ctx.read_changes(self.transactions)

            affected_ids = set()
            affected_ids.update(c.id for c in changed_customers)
            affected_ids.update(t.customer_id for t in new_transactions)

            for cid in affected_ids:
                customer = ctx.read_entity("Customer", cid)
                transactions = ctx.read_related("Transaction", customer_id=cid)
                risk_score = self._calculate_risk(customer, transactions)
                ctx.update(self.risk_scores, cid, {
                    'risk_level': self._score_to_level(risk_score),
                    'risk_score': risk_score,
                })

Incremental Detection Mechanism:

Based on Iceberg snapshot differences:
  Last execution snapshot: snap-1001
  Current snapshot: snap-1005

  Change detection:
    Files added/modified/deleted between snap-1001 and snap-1005
    -> Extract changed entity_id set
    -> Process only those entity_ids

Effect:
  Full execution: 100,000 customers -> 5 minutes
  Incremental: 500 changed customers -> 3 seconds
  Speedup: 100x

class TransformOutputValidator:
    def validate(self, ctx: TransformContext) -> ValidationResult:
        errors = []

        for output in ctx.outputs:
            errors.extend(self._check_schema(output))

        for rel in ctx.new_relationships:
            if not ctx.entity_exists(rel.source_id):
                errors.append(f"Source entity {rel.source_id} not found")
            if not ctx.entity_exists(rel.target_id):
                errors.append(f"Target entity {rel.target_id} not found")

        change_ratio = ctx.change_count / ctx.total_entities
        if change_ratio > 0.5:
            errors.append(
                f"Change ratio {change_ratio:.1%} exceeds 50% safety threshold"
            )

        return ValidationResult(passed=len(errors) == 0, errors=errors)

World Transform Performance Optimization:

1. Parallel sharding
   Partition entities by ID range, execute shards in parallel
   10 shards x 10,000 entities = 10x speedup

2. Vectorized computation
   Use Arrow/NumPy for batch calculations
   Row-by-row -> batch vector = 5-10x speedup

3. Read caching
   Cache frequently accessed entities and relationships
   Avoid repeated Doris queries

4. Incremental execution
   Process only changed data = 10-100x speedup

Combined effect:
  Full unoptimized: 30 minutes
  Full optimized: 3 minutes
  Incremental optimized: 3 seconds

class TestWorldTransform:

    async def test_atomic_rollback_on_failure(self):
        initial = await get_world_state()
        with pytest.raises(TransformError):
            await execute_transform(FailingTransform())
        final = await get_world_state()
        assert initial == final

    async def test_incremental_processes_only_changes(self):
        await execute_transform(RiskScoringIncremental())
        for i in range(5):
            await update_entity(f"customer-{i}", {"credit_rating": "B"})
        result = await execute_transform(RiskScoringIncremental())
        assert result.processed_count == 5

    async def test_concurrent_conflict_detection(self):
        t1 = asyncio.create_task(execute_transform(RiskScoring()))
        t2 = asyncio.create_task(execute_transform(Segmentation()))
        results = await asyncio.gather(t1, t2, return_exceptions=True)
        successes = [r for r in results if not isinstance(r, Exception)]
        assert len(successes) >= 1

    async def test_validation_catches_bad_output(self):
        result = await execute_transform(BadRiskScoring())
        assert result.status == "FAILED"