One Engine for Query, Batch, and Streaming — Ontul’s Exchange Manager and Exactly-Once Semantics

Why Run Three Separate Systems?

Most data platforms handle three workloads with three different systems:

• Interactive SQL: Trino, Presto
• Batch ETL: Spark
• Streaming: Flink, Spark Structured Streaming

Three clusters, three security policies, three monitoring dashboards, three sets of operational overhead. Data gets copied between systems, schemas drift out of sync, and failure points multiply.

Ontul unifies all three into a single engine. Same cluster, same IAM, same connectors, same Admin UI. At the core of this unification is the Exchange Manager.

Exchange Manager: Unified Fault-Tolerance Infrastructure

The Exchange Manager is Ontul’s single infrastructure for fault tolerance across all execution paths.

Query and Batch — Data Spill

Large JOIN, GROUP BY, and ORDER BY queries can exceed available memory. Traditional systems fail with Out of Memory errors.

Ontul’s Exchange Manager detects memory pressure and spills intermediate Arrow RecordBatches to local disk. Sort, HashJoin, HashAggregate — all memory-intensive operators share this path.

Worker (ExecutionEngine)
  └── Sort / HashJoin / HashAggregate
       └── SpillStore → Exchange Manager
            └── KMS envelope encryption → disk write
            └── readback → return to operator

Interactive SQL queries, batch ETL jobs — all flow through the same Exchange Manager. Operators configure a single setting: ontul.exchange.base.dir.

Streaming — State Checkpoint

For streaming, the Exchange Manager serves a different purpose. It persists Kafka consumer offsets and window aggregation state as checkpoints, enabling exact recovery from the point of failure.

Worker (StreamingJobExecutor)
  └── Barrier checkpoint trigger (from Master)
       └── Flush sink → Commit transaction
       └── Snapshot state → Exchange Manager
            └── KMS envelope encryption → disk write
       └── Commit Kafka offsets
       └── Report CHECKPOINT_COMPLETE → Master

One system, one fault-tolerance infrastructure. Data spill and state checkpoint use the same Exchange Manager, encrypted with the same KMS keys, stored in the same directory structure.

Flink-style Continuous Processing

Ontul’s streaming engine is NOT Spark Structured Streaming’s micro-batch approach. It processes events as they arrive, just like Flink — continuous processing.

Records are polled from Kafka (100ms intervals), passed through filter/transform/window aggregation, and written to sinks. The entire loop runs continuously without interruption.

Multi-Worker Distribution

Streaming jobs are distributed across multiple Workers. Partitions are automatically assigned via Kafka consumer groups. When GROUP BY is specified, records are hash-shuffled to the correct Worker via the STREAM_SHUFFLE NIO opcode.

Master
├── CheckpointCoordinator
│   └── CHECKPOINT_TRIGGER every 10s → all Workers
│
Worker-1 (partitions 0-4)          Worker-2 (partitions 5-9)
├── KafkaStreamSource              ├── KafkaStreamSource
├── FILTER                         ├── FILTER
├── WINDOW + GROUP_BY              ├── WINDOW + GROUP_BY
│   └── hash shuffle ─────────────────── hash shuffle
├── AGG (cat-1, cat-3)             ├── AGG (cat-2)
├── Sink (Iceberg/NeorunBase/...)  ├── Sink
└── Exchange Manager checkpoint    └── Exchange Manager checkpoint

Exactly-Once Semantics

The hardest problem in streaming is guaranteeing exactly-once processing. Ontul achieves this by combining barrier checkpoint with transactional sink commits.

Checkpoint Flow

1. Master → CHECKPOINT_TRIGGER → all Workers
2. Each Worker:
   a. Flush shuffle buffers
   b. Flush sink → COMMIT sink transaction
   c. Snapshot state (offsets + window) → Exchange Manager
   d. COMMIT Kafka consumer offsets
   e. CHECKPOINT_COMPLETE → Master
3. Master: all Workers acked → checkpoint globally complete

The key is ordering:

• Sink commit happens first
• Kafka offset commit happens after sink commit
• If crash occurs after sink commit but before offset commit → offsets haven’t advanced, so records will be re-read. But the sink already committed, and transactional isolation prevents duplicates.

Real Pipeline Example

Kafka to Windowed Aggregation to NeorunBase

REST API:

{
  "name": "realtime-agg",
  "type": "STREAMING",
  "config": {
    "source.type": "kafka",
    "source.kafka.bootstrap.servers": "kafka:9092",
    "source.kafka.topic": "user-events",
    "source.kafka.group.id": "ontul-agg",
    "source.kafka.auto.offset.reset": "earliest",
    "sink.type": "neorunbase",
    "sink.mode": "rest",
    "sink.endpoint": "http://neorunbase:8080",
    "sink.tableName": "event_summary"
  },
  "operations": [
    {"type": "FILTER", "value": "amount > 100"},
    {"type": "WINDOW", "value": "TUMBLING(SIZE 10 SECONDS)"},
    {"type": "GROUP_BY", "value": "category"},
    {"type": "AGG", "value": "SUM(amount) as total, COUNT(*) as cnt"}
  ]
}

Ontul SDK in Java:

OntulSession session = OntulSession.builder()
      .master("ontul-master", 47470)           
      .token("jwt-token")           
      .build();          

  String jobId = session.streamSource(
          Source.kafka("inline", "user-events")
              .property("bootstrap.servers", "kafka:9092")
              .groupId("ontul-agg")                       
              .property("auto.offset.reset", "earliest"))
      .filter("amount > 100")                                                                                                                                                                 
      .window(StreamDataFrame.WindowSpec.tumbling(Duration.ofSeconds(10)), "event_time")
      .groupBy("category")                                                                                                                                                                    
      .agg("SUM(amount) as total", "COUNT(*) as cnt")                                                                                                                                         
      .sink(Sink.neorunBase("http://neorunbase:8080", "event_summary")
          .username("admin")                                                                                                                                                                  
          .password("Admin123!")                                                                                                                                                              
          .batchSize(500))      
      .commitInterval(10_000)                                                                                                                                                                 
      .start();                                                                                                                                                                               

  System.out.println("Job started: " + jobId);                                                                                                                                                                                                                                                                                                                            

Ontul SDK in Python:

from ontul.session import OntulSession                                                                                                                                                      

  session = OntulSession(host="ontul-master", port=47470, token="jwt-token")

  job_id = (session.stream_source("user-events",
                bootstrap_servers="kafka:9092",                                                                                                                                               
                group_id="ontul-agg",                       
                auto_offset_reset="earliest")                                                                                                                                                 
      .filter("amount > 100")                               
      .window("TUMBLING(SIZE 10 SECONDS)", event_time_field="event_time")
      .group_by("category")                                                                                                                                                                   
      .agg("SUM(amount) as total", "COUNT(*) as cnt")
      .sink_neorunbase("http://neorunbase:8080", "event_summary",                                                                                                                             
                       username="admin", password="Admin123!")                                                                                                                                
      .commit_interval(10000)                                 
      .start())                                                                                                                                                                               

  print(f"Job started: {job_id}")

This single job:

1. Continuously consumes events from Kafka in real-time
2. Filters to events where amount > 100
3. Aggregates by category in 10-second tumbling windows
4. Writes results to NeorunBase via high-throughput REST bulk-insert
5. Recovers from failure using barrier checkpoint

Query the Same Cluster with SQL

-- Query real-time aggregation results stored in NeorunBase
SELECT * FROM neorun.public.event_summary
WHERE total > 1000 ORDER BY cnt DESC;

-- Cross-engine JOIN with Iceberg CDC data
SELECT n.category, n.total, i.historical_avg
FROM neorun.public.event_summary n
JOIN iceberg.warehouse.category_stats i ON n.category = i.category;

Streaming, Batch, Query — all on the same Ontul cluster, same catalogs, same SQL.

Conclusion

Ontul processes Query, Batch, and Streaming with one engine. At the center is the Exchange Manager.

• Query/Batch: Safe memory spill when operators exceed limits, all data KMS-encrypted
• Streaming: Flink-style continuous processing, barrier checkpoint, exactly-once for transactional sinks
• All paths: Same Exchange Manager, same KMS, same infrastructure

You don’t need three systems. One Ontul cluster is enough.

Reference: https://cloudcheflabs.github.io/ontul-docs