Data Flow Panorama: A Data Point's Complete Journey from Ingestion to Decision
To truly understand a platform's architecture, the best approach is not to study static architecture diagrams, but to trace a single data point through its complete lifecycle.
CoomiaPublished on July 2, 202515 min read
Share this articleTwitter / X
Data Flow Panorama: A Data Point's Complete Journey from Ingestion to Decision
“Series: S2 Architecture Overview · Article 9 | Level: Intermediate | Reading Time: 18 min
#TL;DR
- A single data change in the coomia-dip platform traverses 12 stages from external database to final decision execution, involving 3 processes (onto-control, onto-data, onto-intelligence) and 4 middleware systems (Kafka, Doris, Iceberg, Temporal).
- The data flow uses an event-driven architecture with Kafka decoupling each stage, ensuring that failure at any stage does not block upstream processing, providing system resilience and recoverability.
- The end-to-end latency target from data change to decision execution is < 5 seconds (P99): CDC capture < 1s, Kafka transport < 100ms, rule evaluation < 500ms, decision execution < 2s.
#Introduction: Following a Data Point's Complete Journey
To truly understand a platform's architecture, the best approach is not to study static architecture diagrams, but to trace a single data point through its complete lifecycle.
Consider this scenario: in a supply chain management system, a warehouse's inventory drops from 1000 to 50. This change triggers a cascade of reactions:
- The inventory change is captured
- Data is synced to the platform
- The Ontology object is updated
- Derived properties (inventory turnover rate) are automatically recalculated
- The rule engine detects "inventory below safety threshold"
- The decision engine selects the optimal replenishment plan
- The action engine creates a purchase order
- Audit logs record the entire process
This article traces this data point step by step, from source to destination, showing the processing logic at every layer of the coomia-dip platform.
#1. End-to-End Architecture Diagram
Code
Data Flow Panorama
==================
+---------------+
| External DB | Stage 1: Source data change
| (MySQL/PG) |
+-------+-------+
| binlog/WAL
v
+---------------+
| Flink CDC | Stage 2: Change capture
| Connector |
+-------+-------+
| CDC Event
v
+---------------+
| Kafka | Stage 3: Event transport
| (Topic: |
| cdc.raw) |
+-------+-------+
|
+----+----+
v v
+--------+ +--------+
| Doris | |Iceberg | Stage 4: Data storage
| (real- | |(hist- |
| time) | | ory) |
+----+---+ +--------+
|
v
+--------------+
|onto-data | Stage 5: Ontology Runtime
|(Object Store) | object instance update
+-------+------+
| change event
v
+--------------+
| Kafka | Stage 6: Change subscription
| (Topic: |
| onto.changes)|
+-------+------+
|
+----+----+-----------+
v v v
+------+ +------+ +------+
|Derived| |Sub- | |Mater-| Stage 7: Cascade processing
|Props | |scrip-| |ialize|
|Recalc | |tion | |View |
+--+---+ +------+ +------+
|
v
+--------------+
|onto-intelli | Stage 8: Rule evaluation
|(Rule Engine) |
+-------+------+
| rule match
v
+--------------+
|Decision | Stage 9: Decision selection
|Engine |
+-------+------+
| decision result
v
+--------------+
|Action Engine | Stage 10: Action execution
|(Temporal) |
+-------+------+
|
+----+----+
v v
+------+ +------+
|Ext. | |Audit | Stage 11-12: Execute + Audit
|System| |Log |
+------+ +------+
#2. Stages 1-2: Source Data Change and CDC Capture
#2.1 Scenario Setup
SQL
-- Inventory change in external system
UPDATE inventory SET quantity = 50
WHERE warehouse_id = 'WH-001'
AND product_id = 'PROD-ABC';
-- This UPDATE generates a binlog record:
-- {
-- "table": "inventory",
-- "type": "UPDATE",
-- "before": {"warehouse_id": "WH-001", "product_id": "PROD-ABC", "quantity": 1000},
-- "after": {"warehouse_id": "WH-001", "product_id": "PROD-ABC", "quantity": 50},
-- "ts_ms": 1711234567000
-- }
#2.2 Flink CDC Capture
Flink CDC Connector monitors the external database's binlog (MySQL) or WAL (PostgreSQL), normalizing change events into a unified format:
Java
// Flink CDC Job configuration
public class InventoryCDCJob {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
MySqlSource<String> source = MySqlSource.<String>builder()
.hostname("external-db.example.com")
.port(3306)
.databaseList("supply_chain")
.tableList("supply_chain.inventory")
.username("cdc_reader")
.password("***")
.deserializer(new JsonDebeziumDeserializationSchema())
.build();
env.fromSource(source,
WatermarkStrategy.noWatermarks(),
"inventory-cdc")
.map(new CDCEventTransformer())
.sinkTo(KafkaSink.<String>builder()
.setBootstrapServers("kafka:9092")
.setRecordSerializer(
KafkaRecordSerializationSchema.builder()
.setTopic("cdc.raw.inventory")
.setValueSerializationSchema(
new SimpleStringSchema())
.build())
.build());
env.execute("Inventory CDC Pipeline");
}
}
#2.3 CDC Event Normalization
Raw Debezium events are transformed into the platform's standard CDC Event format:
JSON
{
"event_id": "evt-cdc-001-1711234567",
"event_type": "CDC_UPDATE",
"source": {
"connector": "mysql-inventory",
"database": "supply_chain",
"table": "inventory",
"server_id": "db-001"
},
"before": {
"warehouse_id": "WH-001",
"product_id": "PROD-ABC",
"quantity": 1000
},
"after": {
"warehouse_id": "WH-001",
"product_id": "PROD-ABC",
"quantity": 50
},
"op": "u",
"ts_ms": 1711234567000,
"ontology_mapping": {
"object_type": "InventoryRecord",
"primary_key": "WH-001::PROD-ABC",
"world_id": "world-prod"
}
}
#3. Stages 3-4: Kafka Transport and Data Storage
#3.1 Kafka Topic Design
Code
Kafka Topic Architecture (Three-Tier Topics):
Tier 1: Raw CDC Events
cdc.raw.inventory -- Inventory CDC
cdc.raw.orders -- Order CDC
cdc.raw.shipments -- Shipment CDC
Tier 2: Ontology Change Events
onto.changes.objects -- Object instance changes
onto.changes.links -- Relationship changes
onto.changes.properties -- Property changes
Tier 3: Domain Events
domain.alerts -- Alert events
domain.decisions -- Decision events
domain.actions -- Action execution events
Partition Strategy:
cdc.raw.* -- Hash by primary_key (preserves entity ordering)
onto.changes.* -- Hash by world_id + object_type
domain.* -- Hash by event_type
#3.2 Dual Write: Doris (Real-time) + Iceberg (Historical)
Code
Dual Write Architecture:
Kafka Consumer Group 1: Kafka Consumer Group 2:
onto-data (Doris Writer) onto-data (Iceberg Writer)
| |
v v
+-------------+ +-------------+
| Doris | | Iceberg |
| | | (on Nessie) |
| - Real-time | | |
| queries | | - Historical |
| - Latest | | versions |
| state | | - Time travel|
| - Aggregation| | - Audit trail|
| - ms latency | | |
+-------------+ +-------------+
Write Modes:
Doris: Upsert (update to latest value)
Iceberg: Append (append history) + Nessie version tags
#3.3 Doris Write
Java
// onto-data Doris Writer
@Service
public class DorisObjectWriter {
@Autowired
private DorisTemplate dorisTemplate;
public void upsert(CDCEvent event) {
String tableName = resolveTable(event);
Map<String, Object> row = new HashMap<>();
row.put("pk", event.getPrimaryKey());
row.put("world_id", event.getWorldId());
for (Map.Entry<String, Object> entry :
event.getAfter().entrySet()) {
String columnName = toDorisColumn(
event.getObjectType(), entry.getKey()
);
row.put(columnName, entry.getValue());
}
row.put("_updated_at", event.getTimestamp());
row.put("_version", event.getVersion());
dorisTemplate.upsert(tableName, row);
}
}
#3.4 Iceberg + Nessie Write
Java
// onto-data Iceberg Writer with Nessie version control
@Service
public class IcebergObjectWriter {
@Autowired
private NessieCatalog catalog;
public void appendWithVersion(CDCEvent event) {
TableIdentifier tableId = TableIdentifier.of(
event.getWorldId(),
event.getObjectType()
);
Table table = catalog.loadTable(tableId);
GenericRecord record = GenericRecord.create(table.schema());
record.setField("pk", event.getPrimaryKey());
record.setField("op", event.getOperation());
record.setField("before_json", toJson(event.getBefore()));
record.setField("after_json", toJson(event.getAfter()));
record.setField("ts_ms", event.getTimestamp());
record.setField("event_id", event.getEventId());
DataFile dataFile = writeToParquet(table, record);
table.newAppend()
.appendFile(dataFile)
.commit();
if (shouldTagVersion(event)) {
catalog.commitMeta()
.tag("v-" + event.getTimestamp())
.message("CDC: " + event.getObjectType() + " updated")
.commit();
}
}
}
#4. Stages 5-6: Ontology Runtime and Change Subscription
#4.1 Object Instance Update
After CDC data is written to Doris, onto-data's Ontology Runtime layer maps it to Ontology object instances:
Java
// Ontology Runtime object update
@Service
public class OntologyRuntimeService {
@Autowired
private ObjectTypeRepository objectTypeRepo;
@Autowired
private ChangeEventPublisher eventPublisher;
public void processObjectUpdate(CDCEvent cdcEvent) {
// 1. Find ObjectType definition
ObjectType objectType = objectTypeRepo
.findByApiName(cdcEvent.getObjectType());
// 2. Validate property mapping
Map<String, Object> properties =
mapProperties(objectType, cdcEvent.getAfter());
// 3. Build Ontology change event
ObjectChangeEvent changeEvent = ObjectChangeEvent.builder()
.eventId(generateEventId())
.worldId(cdcEvent.getWorldId())
.objectType(objectType.getApiName())
.objectPrimaryKey(cdcEvent.getPrimaryKey())
.changeType(ChangeType.UPDATE)
.beforeProperties(
mapProperties(objectType, cdcEvent.getBefore()))
.afterProperties(properties)
.changedProperties(
detectChangedProperties(
cdcEvent.getBefore(), cdcEvent.getAfter()))
.sourceEventId(cdcEvent.getEventId())
.timestamp(Instant.now())
.build();
// 4. Publish to Kafka onto.changes.objects
eventPublisher.publish("onto.changes.objects", changeEvent);
}
}
#4.2 Change Event Format
JSON
{
"event_id": "onto-chg-001",
"world_id": "world-prod",
"object_type": "InventoryRecord",
"object_primary_key": "WH-001::PROD-ABC",
"change_type": "UPDATE",
"changed_properties": ["quantity"],
"before_properties": {
"warehouseId": "WH-001",
"productId": "PROD-ABC",
"quantity": 1000
},
"after_properties": {
"warehouseId": "WH-001",
"productId": "PROD-ABC",
"quantity": 50
},
"source_event_id": "evt-cdc-001-1711234567",
"timestamp": "2026-03-24T10:30:01Z"
}
#4.3 Subscription Mechanism
After change events are published to Kafka, multiple consumers subscribe simultaneously:
Code
onto.changes.objects Topic Consumers:
Consumer Group 1: derived-property-calculator
-> Check if any derived properties depend on the changed property
-> If so, trigger derived property recalculation
Consumer Group 2: subscription-notifier
-> Check if any users/applications subscribe to this object type
-> If so, send notifications (WebSocket/Webhook)
Consumer Group 3: materialized-view-updater
-> Check if any materialized views contain this object
-> If so, trigger incremental update
Consumer Group 4: rule-engine-trigger
-> Check if any rules monitor this property change
-> If so, trigger rule evaluation
#5. Stage 7: Derived Property Cascade Recalculation
#5.1 Derived Property Dependency DAG
When the quantity property changes, we need to check which derived properties depend on it:
Code
Derived Property Dependency Graph (DAG):
quantity (base property)
|
+---> inventoryValue (derived)
| = quantity * unitPrice
| |
| +---> totalWarehouseValue (derived)
| = SUM(inventoryValue) GROUP BY warehouseId
|
+---> stockLevel (derived)
= CASE WHEN quantity < safetyThreshold
THEN 'CRITICAL'
WHEN quantity < reorderPoint
THEN 'LOW'
ELSE 'NORMAL' END
|
+---> needsReorder (derived)
= stockLevel IN ('CRITICAL', 'LOW')
#5.2 DAG Traversal and Recalculation
Python
# onto-intelligence derived property recalculation service
class DerivedPropertyCalculator:
"""Derived property cascade recalculation"""
def __init__(self, dag_store, property_store):
self.dag_store = dag_store
self.property_store = property_store
async def on_property_change(
self, change_event: ObjectChangeEvent
):
"""Handle property change, trigger derived recalculation"""
for changed_prop in change_event.changed_properties:
# 1. Find derived properties depending on this one (topo-sorted)
dependents = self.dag_store.get_dependents_topo_sorted(
object_type=change_event.object_type,
property_name=changed_prop,
)
# 2. Recalculate in topological order
for derived_prop in dependents:
new_value = await self._calculate(
derived_prop,
change_event.object_primary_key,
change_event.after_properties,
)
# 3. Update derived property value
await self.property_store.update_derived(
world_id=change_event.world_id,
object_type=change_event.object_type,
object_pk=change_event.object_primary_key,
property_name=derived_prop.name,
value=new_value,
)
# 4. Publish derived change event (may trigger next cascade)
await self._publish_derived_change(
change_event, derived_prop.name, new_value
)
#5.3 Recalculation Results
Code
Recalculation Chain:
quantity: 1000 -> 50
inventoryValue: 1000 * $25 = $25,000 -> 50 * $25 = $1,250
stockLevel: 'NORMAL' -> 'CRITICAL' (safetyThreshold = 100)
needsReorder: false -> true
totalWarehouseValue: $500,000 -> $476,250
#6. Stages 8-9: Rule Evaluation and Decision Selection
#6.1 Rule Engine Trigger
When stockLevel changes to CRITICAL, the rule engine is triggered:
Python
# Rule definition (stored in Ontology)
rule_definition = {
"rule_id": "RULE-INV-001",
"name": "Low inventory auto-reorder rule",
"trigger": {
"object_type": "InventoryRecord",
"property": "stockLevel",
"condition": "value == 'CRITICAL'"
},
"conditions": [
{
"type": "property_check",
"object_type": "InventoryRecord",
"property": "autoReorderEnabled",
"operator": "==",
"value": True
},
{
"type": "time_check",
"constraint": "business_hours"
}
],
"actions": [
{
"type": "trigger_decision",
"decision_id": "DEC-REORDER-001"
}
]
}
#6.2 Rule Evaluation Process
Python
class RuleEvaluationEngine:
"""Rule evaluation engine"""
async def evaluate(
self, change_event: ObjectChangeEvent
) -> list[RuleMatch]:
# 1. Find rules monitoring this property change
candidate_rules = await self.rule_store.find_rules(
object_type=change_event.object_type,
changed_properties=change_event.changed_properties,
)
matches = []
for rule in candidate_rules:
# 2. Check trigger condition
if not self._check_trigger(rule.trigger, change_event):
continue
# 3. Check additional conditions
all_met = True
for condition in rule.conditions:
if not await self._evaluate_condition(
condition, change_event
):
all_met = False
break
if all_met:
matches.append(RuleMatch(
rule=rule,
trigger_event=change_event,
matched_at=datetime.utcnow(),
))
# 4. Resolve rule conflicts (priority ordering)
resolved = self._resolve_conflicts(matches)
# 5. Execute rule actions
for match in resolved:
await self._execute_actions(match)
return resolved
#6.3 Decision Engine
After rule matching, the decision engine selects the optimal replenishment plan:
Python
class DecisionEngine:
"""Decision engine"""
async def execute_decision(
self, decision_id: str, context: dict,
) -> DecisionResult:
decision = await self.decision_store.get(decision_id)
if decision.type == "DECISION_TREE":
return await self._evaluate_tree(decision.tree, context)
elif decision.type == "SCORING_MODEL":
return await self._evaluate_scoring(decision.model, context)
async def _evaluate_tree(self, tree, context):
node = tree.root
while not node.is_leaf:
value = await self._resolve_value(node.property_ref, context)
for branch in node.branches:
if branch.condition.evaluate(value):
node = branch.target
break
return DecisionResult(
action=node.action,
confidence=node.confidence,
reasoning_path=self._get_path(tree.root, node),
)
Decision result example:
JSON
{
"decision_id": "DEC-REORDER-001",
"result": {
"action": "CREATE_PURCHASE_ORDER",
"parameters": {
"supplier_id": "SUP-BEST-001",
"product_id": "PROD-ABC",
"quantity": 500,
"priority": "HIGH",
"delivery_method": "EXPRESS"
},
"confidence": 0.92,
"reasoning_path": [
"stockLevel == CRITICAL -> urgent reorder needed",
"supplier_SUP-BEST-001.leadTime < 3days -> fastest supplier",
"quantity = safetyThreshold * 5 = 500 -> restock to 5x safety"
]
}
}
#7. Stages 10-11: Action Execution
#7.1 Action Engine (Temporal Workflow)
Decision results are transformed into Action execution via Temporal workflow engine for reliable execution:
Python
@workflow.defn
class PurchaseOrderWorkflow:
"""Purchase order creation workflow"""
@workflow.run
async def run(self, params: PurchaseOrderParams) -> ActionResult:
# Step 1: Approval check (amount > $100k needs manual approval)
if params.estimated_cost > 100000:
approval = await workflow.execute_activity(
request_approval,
args=[params],
start_to_close_timeout=timedelta(hours=24),
)
if not approval.approved:
return ActionResult(
status="REJECTED", reason=approval.reason)
# Step 2: Create PO in external ERP
try:
po_result = await workflow.execute_activity(
create_purchase_order_in_erp,
args=[params],
start_to_close_timeout=timedelta(minutes=5),
retry_policy=RetryPolicy(
maximum_attempts=3,
initial_interval=timedelta(seconds=1),
backoff_coefficient=2.0,
),
)
except Exception as e:
await workflow.execute_activity(
notify_po_creation_failure,
args=[params, str(e)],
start_to_close_timeout=timedelta(minutes=1),
)
raise
# Step 3: Update Ontology object status
await workflow.execute_activity(
update_inventory_reorder_status,
args=[params.product_id, po_result.po_id],
start_to_close_timeout=timedelta(minutes=1),
)
# Step 4: Send notification
await workflow.execute_activity(
send_notification,
args=[
f"Purchase order {po_result.po_id} created",
params.notify_users,
],
start_to_close_timeout=timedelta(minutes=1),
)
return ActionResult(
status="COMPLETED",
po_id=po_result.po_id,
details=po_result,
)
#7.2 Action Execution Timeline
Code
Action Execution Timeline:
Time onto-intelligence Temporal External ERP
|
| Decision result ------> Start workflow
| |
| Approval check
| (auto-approved,
| amount < $100k)
| |
| Activity:
| create_po ----------> POST /api/po
| | |
| | Created OK
| | <---------------- PO-2026-001
| |
| Activity:
| update_onto
| <--------------------- gRPC Update
| Update object status
| |
| Activity: notify
| |
| Workflow complete
#8. Stage 12: Audit Logging
#8.1 Full Chain Audit
Every stage produces audit records, aggregated into the audit log service:
Python
class AuditLogger:
"""Full chain audit logging"""
async def log_full_chain(self, chain_id: str, events: list):
audit_record = {
"chain_id": chain_id,
"stages": [
{
"stage": "CDC_CAPTURE",
"timestamp": "2026-03-24T10:30:00.100Z",
"source": "mysql-inventory",
"duration_ms": 50,
},
{
"stage": "KAFKA_TRANSPORT",
"timestamp": "2026-03-24T10:30:00.150Z",
"topic": "cdc.raw.inventory",
"duration_ms": 80,
},
{
"stage": "DORIS_WRITE",
"timestamp": "2026-03-24T10:30:00.230Z",
"duration_ms": 20,
},
{
"stage": "ONTOLOGY_UPDATE",
"timestamp": "2026-03-24T10:30:00.300Z",
"object_type": "InventoryRecord",
"duration_ms": 100,
},
{
"stage": "DERIVED_PROPERTY_CALC",
"timestamp": "2026-03-24T10:30:00.500Z",
"properties": ["inventoryValue", "stockLevel",
"needsReorder"],
"duration_ms": 200,
},
{
"stage": "RULE_EVALUATION",
"timestamp": "2026-03-24T10:30:00.800Z",
"rule_id": "RULE-INV-001",
"result": "MATCHED",
"duration_ms": 150,
},
{
"stage": "DECISION_EXECUTION",
"timestamp": "2026-03-24T10:30:01.000Z",
"decision_id": "DEC-REORDER-001",
"confidence": 0.92,
"duration_ms": 300,
},
{
"stage": "ACTION_EXECUTION",
"timestamp": "2026-03-24T10:30:01.500Z",
"action": "PurchaseOrderWorkflow",
"result": "COMPLETED",
"po_id": "PO-2026-001",
"duration_ms": 2500,
},
],
}
await self.audit_store.save(audit_record)
#8.2 End-to-End Latency Analysis
Code
End-to-End Latency Breakdown (P99 Targets):
Stage Target Actual
---------------------------------------------
CDC Capture < 1000ms ~100ms
Kafka Transport < 100ms ~80ms
Doris Write < 50ms ~20ms
Ontology Update < 200ms ~100ms
Derived Property Recalc < 500ms ~200ms
Rule Evaluation < 500ms ~150ms
Decision Execution < 500ms ~300ms
Action Execution < 3000ms ~2500ms
---------------------------------------------
Total < 5850ms ~3450ms
P99 End-to-End Latency Target: < 5 seconds
Actual P99 Latency: ~3.5 seconds
#9. Failure Scenarios
#9.1 Failure Handling by Stage
Code
Failure Handling Matrix:
Stage Failure Type Strategy
---------------------------------------------------------
CDC Capture DB connection lost Auto-reconnect + resume from binlog pos
Kafka Transport Broker unavailable Producer buffering + retry
Doris Write Write timeout DLQ + alert + manual replay
Iceberg Write Nessie conflict Optimistic retry (3 attempts)
Ontology Update ObjectType not found Log to DLQ + notify admin
Derived Calc Expression error Mark property ERROR + alert
Rule Evaluation Rule conflict Priority ordering + logging
Decision Exec Model timeout Degrade to default decision
Action Execution External system failure Temporal auto-retry + compensation
Audit Logging Write failure Local buffer + async retry
#9.2 Dead Letter Queue (DLQ) Design
Code
DLQ Architecture:
Normal flow:
cdc.raw.* ---> Consumer ---> Success ---> Next stage
Error flow:
cdc.raw.* ---> Consumer ---> Failure
|
v (after 3 retries)
dlq.cdc.raw.*
|
v
DLQ Monitor
(scans every minute)
|
+----------+----------+
v v
Auto-replay Alert + manual
(recoverable) (non-recoverable)
#10. Performance Optimization
#10.1 Batch Processing
Code
Batch Optimization:
Scenario: Large volume CDC events from same table (bulk inventory update)
Before optimization: Process one by one
CDC Event 1 -> Kafka -> Doris Write -> Onto Update -> ...
CDC Event 2 -> Kafka -> Doris Write -> Onto Update -> ...
1000 events * 3.5s = 3500s (unacceptable)
After optimization: Micro-batch processing
CDC Events 1-100 -> Kafka (batch) -> Doris Batch Write ->
Onto Batch Update -> Derived Properties Batch Recalc
Batch size: 100 events/batch
Batch window: 500ms max
1000 events = 10 batches * 500ms = 5s (acceptable)
#10.2 Derived Property Recalculation Optimization
Code
Optimization Strategies:
1. Change coalescing: Merge multiple changes to same object within 500ms window
2. DAG pruning: Only recalculate actually affected derived property branches
3. Cache intermediate results: Cache intermediate computations for same DAG paths
4. Parallel computation: Recalculate independent derived properties in parallel
#Key Takeaways
-
Event-driven architecture is the optimal choice for data-intensive platforms: Kafka decouples 12 processing stages, each scaling and failing independently, ensuring system resilience and observability.
-
Ontology is the semantic hub of data flow: Raw CDC events gain business semantics after Ontology mapping; subsequent derived properties, rules, and decisions all operate on Ontology semantics rather than raw data fields.
-
End-to-end traceability is the lifeline of intelligent decision platforms: Every step from data change to decision execution has an audit record, with chain_id linking the complete chain — satisfying both compliance requirements and debugging needs.
#Next Article Preview
S2-10 Consistency Model: Data Consistency Design in a Distributed System — The data flow panorama showed how data flows, but didn't answer a critical question: when multiple Worlds modify concurrently and multiple consumers process in parallel, how do we ensure data consistency? The next article deep-dives into Nessie optimistic concurrency, Kafka event ordering guarantees, World isolation, and consistency without distributed transactions.
tags: Data-Flow, CDC, Kafka, Doris, Iceberg, Ontology-Runtime, Derived-Property, Rule-Engine, Decision-Engine, Temporal, Audit, coomia-dip