Computation Strategy Routing: Priority Scheduling Across 7 Compute Modes

“

Series: S2 Architecture Overview · Article 7 | Level: Intermediate | Reading Time: 18 min

TL;DR

• coomia-dip supports 7 computation modes (cache lookup, materialized views, SQL computation, expression evaluation, reducer aggregation, function compute, and real-time streaming). The ComputationCoordinator selects the optimal execution path for every property query.
• Routing decisions are based on a cost model that considers data freshness requirements, query latency budgets, resource consumption, and cache hit rates, completing path selection in milliseconds.
• This computation routing mechanism is the core engine behind coomia-dip's Derived Property system. It frees business logic from needing to know whether data comes from cache, Doris computation, or a streaming pipeline -- developers simply declare "what I need" and the router decides "how to compute it."

#1. Introduction: Why Computation Routing Matters

In the previous article on event-driven architecture, we explored how Kafka connects platform components. But after an event fires, how is the actual property value computed? That is the problem computation routing solves.

Consider a straightforward scenario: a supply chain platform needs to calculate a "Supplier Composite Risk Score." This derived property depends on:

• Historical on-time delivery rate (requires aggregating millions of delivery records)
• Quality complaints in the last 30 days (requires time-window filtering)
• Financial health index (requires calling an external API)
• Industry peer ranking percentile (requires cross-supplier comparison)

In traditional architectures, performing a full computation on every query is clearly impractical. But naively caching results might return stale data. coomia-dip's ComputationCoordinator is designed to solve exactly this class of problems -- it dynamically selects the most appropriate computation path for each query.

#2. Overview of the Seven Computation Modes

#2.1 Mode Classification

coomia-dip categorizes all property computations into 7 modes, listed by descending priority:

#2.2 Priority Does Not Mean Fixed Order

A crucial concept: priority determines "preference," not "fixed order." The ComputationCoordinator dynamically adjusts the actual selection based on multiple factors:

Routing Decision = f(freshness_requirement, latency_budget, cache_state, resource_availability, data_scale)

For example, even though cache has the highest priority, if the business requires "freshness < 1 second" and the cache was last refreshed 30 seconds ago, the router skips cache and selects SQL or STREAMING mode directly.

#3. ComputationCoordinator Core Architecture

#3.1 Component Positioning

The ComputationCoordinator resides within OntologyRuntimeService in Control Layer (Control Layer), serving as the core scheduler for derived property queries.

┌─────────────────────────────────────────────────┐
│              OntologyRuntimeService             │
│                                                 │
│  ┌─────────────────────────────────────────┐    │
│  │       ComputationCoordinator            │    │
│  │                                         │    │
│  │  ┌───────────┐  ┌──────────────────┐   │    │
│  │  │ CostModel │  │ FreshnessChecker │   │    │
│  │  └───────────┘  └──────────────────┘   │    │
│  │  ┌───────────┐  ┌──────────────────┐   │    │
│  │  │ RouteTable│  │ CircuitBreaker   │   │    │
│  │  └───────────┘  └──────────────────┘   │    │
│  └─────────────────────────────────────────┘    │
│                                                 │
│  ┌──────┐ ┌────────────┐ ┌─────┐ ┌──────────┐ │
│  │Cache │ │Materialized│ │ SQL │ │Expression│ │
│  │Engine│ │  Engine    │ │Exec │ │ Evaluator│ │
│  └──────┘ └────────────┘ └─────┘ └──────────┘ │
│  ┌──────────┐ ┌──────────┐ ┌─────────────────┐│
│  │ Reducer  │ │ Function │ │ StreamingEngine ││
│  │ Engine   │ │ Runtime  │ │                 ││
│  └──────────┘ └──────────┘ └─────────────────┘│
└─────────────────────────────────────────────────┘

#3.2 Routing Decision Flow

Every derived property query proceeds through the following steps:

Step 1: Parse Request
  ├─ Extract ObjectType + PropertyName
  ├─ Read the property's ComputationSpec (computation rule definition)
  └─ Determine freshness_requirement and latency_budget

Step 2: Evaluate Candidate Modes
  ├─ Check each mode's availability (is the engine healthy?)
  ├─ Check cache hit status and freshness
  ├─ Estimate execution cost for each mode
  └─ Filter out modes that violate constraints

Step 3: Select Optimal Path
  ├─ Sort candidate modes by cost
  ├─ Select the lowest-cost mode meeting all constraints
  └─ Record routing decision (for audit and optimization)

Step 4: Execute Computation
  ├─ Dispatch request to the selected compute engine
  ├─ Set timeout and circuit-breaker protection
  └─ Return result and asynchronously update cache

#3.3 Cost Model in Detail

The cost model is the heart of routing decisions. Each computation mode's cost is estimated using the following formula:

def estimate_cost(mode, context):
    """Compute mode cost estimation"""
    base_cost = MODE_BASE_COSTS[mode]          # Base overhead
    data_cost = context.data_size * MODE_DATA_FACTORS[mode]  # Data scale factor
    resource_cost = get_resource_pressure(mode) # Current resource pressure
    freshness_penalty = calc_freshness_gap(     # Freshness penalty
        mode, context.freshness_requirement
    )

    total = base_cost + data_cost + resource_cost + freshness_penalty

    # Cache hits have extremely low cost
    if mode == CACHE and cache_hit(context):
        total *= 0.01

    return total

#4. Mode 1: Cache Lookup (CACHE)

#4.1 Cache Layer Architecture

Cache lookup is the fastest computation path. coomia-dip uses a two-tier cache architecture:

L1 Cache: In-process cache (Caffeine)
  ├─ Capacity: 10,000 entries per service instance
  ├─ Eviction: Auto-expire 30 seconds after write
  └─ Hit latency: < 0.1ms

L2 Cache: Distributed cache (Redis Cluster)
  ├─ Capacity: No hard limit, managed by memory quota
  ├─ Eviction: Per property-defined cache_ttl
  └─ Hit latency: 0.5-2ms

#4.2 Cache Invalidation Strategy

Cache invalidation uses a "event-driven + TTL dual guarantee" approach:

1. Event-driven invalidation: When source data changes, CDC events via Kafka trigger cache invalidation. Latency from data change to cache invalidation is typically 50-200ms.

2. TTL backstop: Even if events are lost, TTL ensures the cache eventually expires. Default TTL varies by property type:

   • Static properties (e.g., name, identifier): TTL = 5 minutes
   • Near-real-time properties (e.g., inventory count): TTL = 30 seconds
   • Real-time properties (e.g., price): TTL = 5 seconds

#4.3 Cache Warming

On system startup, the ComputationCoordinator pre-warms the top-1,000 hot properties based on historical access frequency. Warming runs asynchronously and does not block service startup.

#5. Mode 2: Materialized Views (MATERIALIZED)

#5.1 Materialized Views' Role

Materialized views suit properties that are expensive to compute but change infrequently. For example:

• "Department annual revenue" -- requires aggregating tens of thousands of orders, but only needs daily updates
• "Product average rating" -- requires aggregating all reviews, but hourly updates suffice

#5.2 Doris Materialized Views

coomia-dip leverages Apache Doris's materialized view capabilities for pre-computation:

-- Materialized view (automatically managed by ComputationCoordinator)
CREATE MATERIALIZED VIEW mv_supplier_risk_score AS
SELECT
    supplier_id,
    AVG(delivery_on_time_rate) as avg_delivery_rate,
    COUNT(quality_complaint_id) as complaint_count,
    MAX(last_assessment_date) as latest_assessment
FROM supplier_delivery_records
LEFT JOIN quality_complaints USING (supplier_id)
GROUP BY supplier_id;

#5.3 Materialization Refresh Strategies

Materialized view refreshes are centrally scheduled by the ComputationCoordinator:

#6. Mode 3: SQL Computation

#6.1 Dynamic SQL Generation

For property queries that cannot be pre-computed, the ComputationCoordinator dynamically generates SQL from the property's ComputationSpec:

class SQLComputationEngine:
    def compute(self, spec: ComputationSpec, context: QueryContext):
        # Build SQL from ComputationSpec
        sql_builder = SQLBuilder(spec.object_type)

        # Add computation logic
        for rule in spec.computation_rules:
            sql_builder.add_computation(rule)

        # Add filter conditions
        sql_builder.add_filters(context.filters)

        # Add safety bounds (prevent full table scans)
        sql_builder.add_limit(spec.max_rows or 10000)
        sql_builder.add_timeout(context.latency_budget)

        # Execute query
        return self.doris_client.execute(sql_builder.build())

#6.2 Query Safety Protections

SQL computation mode includes multiple safety mechanisms:

• Row limit: Maximum 10,000 rows by default to prevent memory overflow
• Timeout: Query timeout defaults to 30 seconds with automatic termination
• Resource quota: Each tenant is limited to 20 concurrent SQL queries
• SQL injection protection: All parameters pass through prepared statements; string concatenation is forbidden

#7. Mode 4: Expression Evaluation (EXPRESSION)

#7.1 Expression Engine

Expression evaluation handles simple property combination calculations, for example:

// Full name = last name + " " + first name
full_name = last_name + " " + first_name

// Gross margin = (revenue - cost) / revenue * 100
gross_margin = (revenue - cost) / revenue * 100

// Status label = IF(score > 80, "Excellent", IF(score > 60, "Pass", "Fail"))
status_label = IF(score > 80, "Excellent", IF(score > 60, "Pass", "Fail"))

#7.2 Expression Compilation and Caching

To avoid parsing expression strings on every execution, the engine compiles expressions into ASTs and caches them:

Source Expression ──parse──> AST ──optimize──> Optimized AST ──compile──> Executable Function
                                                                              │
                                                                        Cached in memory
                                                                        (Key = expression hash)

Compiled expression execution latency is typically in the microsecond range, making expression mode the second-fastest computation path after cache.

#7.3 Type Safety

The expression engine performs type checking during the compilation phase:

revenue (Decimal) - cost (Decimal) → Decimal ✓
name (String) + age (Integer) → Type Error ✗

Type-incompatible expressions are rejected when the property definition is saved, rather than throwing exceptions at runtime.

#8. Mode 5: Reducer Aggregation

#8.1 MapReduce-Style Distributed Aggregation

When data scale exceeds single-machine processing capacity, the ComputationCoordinator selects Reducer mode. This mode decomposes aggregation tasks into Map and Reduce phases:

Map Phase (parallel):
  Partition 1 ──> Partial aggregate 1
  Partition 2 ──> Partial aggregate 2
  Partition 3 ──> Partial aggregate 3
  ...
  Partition N ──> Partial aggregate N

Reduce Phase (merge):
  Partial results 1..N ──> Global aggregate result

#8.2 Supported Aggregation Operations

#8.3 SQL vs Reducer Selection Boundary

The ComputationCoordinator automatically chooses between SQL and Reducer based on data volume:

Data volume < 1 million rows → SQL mode (single-node Doris is sufficient)
Data volume 1M - 100M rows  → Reducer mode (distributed aggregation)
Data volume > 100M rows     → Materialized view (pre-compute + incremental update)

#9. Mode 6: Function Compute (FUNCTION)

#9.1 Custom Function Registration

Function compute mode allows developers to register custom computation logic for cases that cannot be expressed with SQL or expressions:

# Register custom function in Reasoning & Decision Layer (Intelligence Layer)
@computation_function(
    name="supplier_risk_score",
    input_types={"supplier_id": "string"},
    output_type="decimal",
    timeout_ms=5000,
    cacheable=True,
    cache_ttl_seconds=3600
)
async def compute_supplier_risk(supplier_id: str) -> Decimal:
    """Compute composite supplier risk score"""
    # Fetch multi-dimensional data
    delivery_data = await get_delivery_history(supplier_id)
    financial_data = await get_financial_health(supplier_id)
    complaint_data = await get_quality_complaints(supplier_id)

    # Multi-dimensional weighted calculation
    score = (
        delivery_data.on_time_rate * 0.35 +
        financial_data.health_index * 0.30 +
        (1 - complaint_data.rate) * 0.20 +
        delivery_data.response_speed * 0.15
    )

    return Decimal(str(round(score, 2)))

#9.2 Function Call Chain

Function compute invocations traverse gRPC cross-Layer communication:

Control Layer (B)                    Intelligence Layer (D)
ComputationCoordinator               FunctionRuntime
       │                                    │
       ├──gRPC──> ExecuteFunction ──────────>│
       │          (supplier_risk_score,      │
       │           {supplier_id: "S001"})    │
       │                                    │
       │                              Computing...
       │                                    │
       │<──gRPC── FunctionResult <──────────│
       │          (score: 0.78,             │
       │           compute_time_ms: 230)    │

#9.3 Function Compute Safeguards

• Timeout control: Each function has an independent timeout setting (default 5 seconds)
• Resource isolation: Functions execute in isolated sandboxes without affecting the main process
• Retry policy: Idempotent functions support automatic retries (up to 3 attempts)
• Circuit breaker: After 5 consecutive failures the circuit opens, with a half-open probe after 60 seconds

#10. Mode 7: Real-time Streaming (STREAMING)

#10.1 Streaming Compute Engine

Real-time streaming is the highest-latency but most real-time computation mode. It applies to:

• Metrics requiring real-time aggregation (e.g., "order count in the last 5 minutes")
• Complex Event Processing (CEP) rules (e.g., "3 consecutive anomaly alerts")
• Real-time risk scoring

#10.2 Integration with Kafka Streams

coomia-dip's streaming compute engine is built on Kafka Streams:

Kafka Topic                Streaming Engine               Result
(source events)                                          (output)
     │                                                      │
     ├─> Window(5min) ─> Count ─> "Orders last 5 min" ────>│
     ├─> Window(1h)  ─> Avg  ─> "Hourly avg price" ───────>│
     ├─> CEP Rule    ─> Match ─> "Anomaly detection" ─────>│
     └─> Tumbling(1d)─> Sum  ─> "Daily cumulative" ───────>│

#10.3 Window Types

#10.4 Streaming Fallback Strategy

When the streaming engine is unavailable, the ComputationCoordinator automatically falls back to SQL mode:

async def compute_with_fallback(spec, context):
    try:
        # Prefer streaming compute
        result = await streaming_engine.compute(spec, context)
        return result
    except StreamingUnavailableError:
        # Fall back to SQL (higher latency but still available)
        logger.warning(
            "Streaming engine unavailable, falling back to SQL",
            property=spec.property_name
        )
        return await sql_engine.compute(spec, context)

#11. Advanced Routing Features

#11.1 Multi-Property Batch Routing

When a single query requests multiple derived properties, the ComputationCoordinator performs batch optimization:

Request: Get supplier S001's [risk_score, on_time_rate, quality_grade, active_status]

Batch routing optimization:
  risk_score    → FUNCTION (requires custom computation)
  on_time_rate  → CACHE (high-frequency access, cache hit)
  quality_grade → MATERIALIZED (pre-computed materialized view)
  active_status → EXPRESSION (simple boolean expression)

Parallel execution: CACHE + EXPRESSION return immediately
                    MATERIALIZED + FUNCTION compute in parallel
Merged result: Final latency = max(all path latencies) ≈ 230ms

#11.2 Adaptive Routing Learning

The ComputationCoordinator records the actual execution results of every routing decision and uses them to adjust cost model parameters:

historical_stats = {
    "supplier_risk_score": {
        "CACHE":        {"avg_latency": 0.5,  "hit_rate": 0.85},
        "SQL":          {"avg_latency": 45,   "success_rate": 0.99},
        "FUNCTION":     {"avg_latency": 230,  "success_rate": 0.97},
        "MATERIALIZED": {"avg_latency": 3,    "freshness_gap": 3600}
    }
}

# Dynamically adjust cost factors based on historical data
cost_factor["SQL"] = base_cost * (1 + failure_rate * penalty)

#11.3 Tenant-Level Routing Policies

Different tenants can configure different routing preferences:

tenant_routing_config:
  tenant_gold:
    # Gold tier: prioritize freshness
    freshness_weight: 0.8
    latency_weight: 0.1
    cost_weight: 0.1
    max_streaming_partitions: 16

  tenant_silver:
    # Silver tier: balanced strategy
    freshness_weight: 0.4
    latency_weight: 0.3
    cost_weight: 0.3
    max_streaming_partitions: 8

  tenant_basic:
    # Basic tier: prioritize cost
    freshness_weight: 0.2
    latency_weight: 0.2
    cost_weight: 0.6
    streaming_enabled: false

#12. Circuit Breaking and Graceful Degradation

#12.1 Per-Engine Circuit Breakers

The ComputationCoordinator maintains independent circuit breaker state for each compute engine:

Circuit Breaker Status:
  CACHE        → CLOSED (healthy)   ── Consecutive failures 0/5
  MATERIALIZED → CLOSED (healthy)   ── Consecutive failures 0/5
  SQL          → HALF_OPEN (probing) ── Last tripped 30s ago
  EXPRESSION   → CLOSED (healthy)   ── Consecutive failures 0/5
  REDUCER      → CLOSED (healthy)   ── Consecutive failures 1/5
  FUNCTION     → OPEN (tripped)     ── Will probe in 60s
  STREAMING    → CLOSED (healthy)   ── Consecutive failures 0/5

#12.2 Graceful Degradation Paths

When higher-priority modes are unavailable, the router automatically selects the next best mode:

Ideal path:       CACHE → hit → return (0.5ms)
Degradation L1:   CACHE → miss → MATERIALIZED → return (3ms)
Degradation L2:   CACHE → miss → MATERIALIZED → stale → SQL → return (45ms)
Degradation L3:   All unavailable → return default value + alert (0ms + alert)

#12.3 Backpressure Control

When system load is high, the ComputationCoordinator proactively restricts expensive computation modes:

class BackpressureController:
    def should_allow(self, mode: ComputationMode) -> bool:
        current_load = get_system_load()

        if current_load > 0.9:  # System load > 90%
            # Only allow CACHE and EXPRESSION
            return mode in (CACHE, EXPRESSION)

        if current_load > 0.7:  # System load > 70%
            # Disable STREAMING and FUNCTION
            return mode not in (STREAMING, FUNCTION)

        return True  # Normal mode, all allowed

#13. Comparison with Palantir Foundry

coomia-dip's core advantage is dynamic routing -- Foundry's computation paths are determined at Pipeline build time, whereas coomia-dip selects the optimal path at each query in real time. This means that when facing data growth, load changes, or component failures, coomia-dip adapts automatically without manual intervention.

#14. Observability and Tuning

#14.1 Routing Decision Logs

Every routing decision is logged in detail for analysis and optimization:

{
  "trace_id": "abc-123",
  "property": "supplier_risk_score",
  "object_id": "S001",
  "candidates": [
    {"mode": "CACHE", "cost": 0.5, "available": true, "fresh": false},
    {"mode": "MATERIALIZED", "cost": 3.2, "available": true, "fresh": true},
    {"mode": "SQL", "cost": 45.0, "available": true, "fresh": true},
    {"mode": "FUNCTION", "cost": 230.0, "available": false, "reason": "circuit_open"}
  ],
  "selected": "MATERIALIZED",
  "reason": "lowest_cost_meeting_freshness",
  "actual_latency_ms": 2.8,
  "timestamp": "2026-03-24T10:15:30Z"
}

#14.2 Key Monitoring Metrics

#14.3 Performance Tuning Recommendations

1. Improve cache hit rate: Analyze cache-miss logs and adjust TTL or expand warming scope for high-frequency properties
2. Materialized view coverage: For frequently executed SQL-mode queries, consider creating materialized views
3. Function compute optimization: For functions exceeding 200ms, investigate whether they can be decomposed into cache + incremental computation
4. Reduce fallbacks: Engines that frequently trip circuit breakers need root-cause analysis (typically resource insufficiency or network issues)

#Key Takeaways

1. Seven computation modes cover the complete latency spectrum from microseconds to seconds. The ComputationCoordinator does not simply "try cache, fall back to database." Instead, it performs globally optimal selection across 7 modes based on a cost model. Each mode has clearly defined use cases, performance characteristics, and protection mechanisms, forming a comprehensive computation strategy matrix.

2. Dynamic routing gives the platform self-adaptive capabilities. Traditional systems fix computation paths at development time, requiring manual adjustment when data volumes grow or loads change. coomia-dip's cost model dynamically adjusts based on real-time metrics and historical statistics, with routing decisions automatically adapting to system state changes. Combined with multi-tenant policy configuration, different business scenarios receive customized computation experiences.

3. Circuit breaking and degradation ensure the system never becomes fully unavailable due to compute engine failures. The 7 computation modes themselves form a natural degradation chain. Even in the worst case where only the expression engine remains available, the system can still return basic computation results. This tiered degradation strategy elevates platform availability from single-point availability to matrix availability.

“

Next Article Preview: [S2-08] API Design Philosophy: Design Principles of Ontology-Native APIs -- a deep dive into how coomia-dip maps ontology concepts to RESTful APIs and gRPC interfaces, delivering an "ontology-centric" API experience.

Tags: #computation-routing #derived-property #cost-model #cache #materialized-view #sql #expression #reducer #function #streaming #circuit-breaker #coomia-dip