How the RegionServer Handles Read and Write Requests in the GreptimeDB Datanode

The RegionServer in GreptimeDB's datanode processes read requests via Arrow Flight by decoding Tickets into logical plans and executing them through the QueryEngine, while write requests arrive via gRPC and are dispatched to RegionEngines (Mito, Metric, etc.) with optional batch optimizations for high-throughput inserts.

The RegionServer serves as the core request handler on each GreptimeDB datanode, exposing two distinct network protocols to clients. It accepts analytical read workloads through the Arrow Flight interface and ingests write operations through a custom Greptime gRPC protocol. Understanding these dual paths is essential for operators tuning concurrency limits and for developers extending the storage engine.

Overview of the RegionServer Architecture

The RegionServer implementation resides primarily in src/datanode/src/region_server.rs and exposes two entry points:

• Read path – Arrow Flight do_get receives a Ticket containing an encoded QueryRequest. The server decodes this into a logical plan, rewrites it with region-aware data sources, and streams results back as Arrow RecordBatches.
• Write path – gRPC RegionServerHandler::handle receives RegionRequest messages (Insert, Delete, Alter, etc.). Requests are routed to the appropriate RegionEngine (Mito, Metric, or others) and executed with parallelism limits to prevent overload.

Both paths share a concurrency limiter (RegionServerParallelism) and maintain region state metadata (Registering, Deregistering, Ready) to ensure consistency during lifecycle operations.

Handling Read Requests via Arrow Flight

The do_get Entry Point

When a client initiates a read, it sends an Arrow Flight Ticket to the do_get method defined in src/datanode/src/region_server.rs at lines 992–1010. This method acts as the primary ingress for all analytical queries.

// src/datanode/src/region_server.rs
async fn do_get(
    &self,
    request: Request<Ticket>,
) -> Result<Response<BoxStream<'static, Result<FlightData, Status>>>> {
    let ticket = request.into_inner();
    let query_request = QueryRequest::decode(&*ticket.ticket)
        .map_err(|e| Status::invalid_argument(e.to_string()))?;
    // ... proceeds to handle_remote_read
}

Decoding and Context Setup

After decoding the protobuf QueryRequest, the server extracts the region ID and builds a QueryContext. It also initializes distributed tracing by parsing W3C trace headers from the request metadata, allowing end-to-end observability across the cluster.

Query Execution Flow

The core logic resides in handle_remote_read (lines 246–291) and handle_read (lines 294–322):

1. Acquire parallelism permit – If RegionServerParallelism is configured, the server acquires a semaphore permit with timeout to bound concurrent reads (line 252).
2. Resolve region and catalog – The server creates a NameAwareCatalogList that injects the region-specific data source, ensuring the query planner targets the correct physical storage (lines 258–264).
3. Decode logical plan – The encoded plan from the request is deserialized using the query engine's plan decoder (lines 269–277).
4. Rewrite and execute – handle_read uses a NameAwareDataSourceInjector to rewrite the plan with the correct table provider (lines 304–312), then executes via QueryEngine (lines 316–318).
5. Stream results – The resulting SendableRecordBatchStream is wrapped in a FlightRecordBatchStream and returned to the client.

Concurrency Control and Suspension

The server implements two protective mechanisms:

• Suspension – If the node is suspended (self.is_suspended()), all read requests immediately return a SuspendedSnafu error, allowing for maintenance or graceful shutdown.
• Parallelism limits – The optional RegionServerParallelism uses a Tokio semaphore to enforce maximum concurrent reads, preventing memory exhaustion under heavy analytical workloads.

Handling Write Requests via gRPC

Request Dispatch and Routing

Write requests enter through the gRPC RegionServerHandler::handle method in src/servers/src/grpc/region_server.rs (lines 51–71). The handler receives a region_request::Body and dispatches based on the variant:

• DDL operations (Create, Drop, Alter) → handle_batch_ddl_requests
• Inserts/Deletes → handle_requests_in_parallel
• Sync → handle_sync_region_request
• ListMetadata → handle_list_metadata_request
• Others → handle_requests_in_serial

Parallel Write Processing

For high-throughput insert and delete workloads, the server uses handle_requests_in_parallel (defined in src/datanode/src/region_server.rs). This method constructs a vector of async tasks, each invoking handle_request for a specific region, and executes them concurrently using try_join_all:

// src/datanode/src/region_server.rs
let join_tasks = requests.into_iter().map(|(region_id, req)| {
    let self_to_move = self.clone();
    let span = tracing_context.attach(info_span!(
        "RegionServer::handle_region_request",
        region_id = region_id.to_string()
    ));
    async move {
        self_to_move
            .handle_request(region_id, req)
            .trace(span)
            .await
    }
});
let results = try_join_all(join_tasks).await?;

Before processing, the server checks region state via get_engine and transitions the region to Registering or Deregistering as necessary to prevent concurrent modifications to region metadata.

MetricEngine Batch Insert Optimization

When all requests in a batch are Put operations targeting the MetricEngine, the server bypasses the generic per-region loop and invokes MetricEngine::put_regions_batch directly (lines 450–470 in region_server.rs):

// src/datanode/src/region_server.rs
let metric_engine = engine
    .as_any()
    .downcast_ref::<MetricEngine>()
    .context(UnexpectedSnafu { violated: "Failed to downcast to MetricEngine" })?;
let result = metric_engine
    .put_regions_batch(put_requests)
    .trace(span)
    .await
    .map_err(BoxedError::new)
    .context(HandleRegionRequestSnafu { region_id: first_region_id });

This optimization yields a single RegionResponse aggregating total_affected rows and updates the REGION_CHANGED_ROW_COUNT metric, significantly reducing overhead for high-cardinality metric ingestion.

Serial Write Processing for DDL Operations

Certain request types—such as Alter, Flush, and Compact—require strict ordering to avoid race conditions. These are handled by handle_requests_in_serial, which processes requests sequentially:

for (region_id, req) in requests {
    let span = tracing_context.attach(info_span!(
        "RegionServer::handle_region_request",
        region_id = region_id.to_string()
    ));
    let result = self.handle_request(region_id, req).trace(span).await?;
    // Aggregate affected_rows and extensions...
}

This ensures that schema modifications or maintenance operations complete atomically without interference from concurrent writes.

Per-Region Request Handling

The internal handle_request method (lines 380–420) performs the final dispatch to the appropriate RegionEngine:

1. Determine region change – Maps the request variant (Register, Deregister, Ingest, etc.) to a RegionChange type.
2. Retrieve engine – Calls self.inner.get_engine(region_id, &region_change) to obtain the correct RegionEngine (Mito, Metric, etc.).
3. State management – Sets the region status to not ready via set_region_status_not_ready before execution.
4. Execution – Invokes engine.handle_request(region_id, request).await.
5. Completion – On success, sets status to ready via set_region_status_ready; on failure, unsets the engine to prevent stale metadata.

Write-specific metrics such as REGION_CHANGED_ROW_COUNT and REGION_SERVER_INSERT_FAIL_COUNT are updated based on the request outcome.

Key Source Files and Implementation Details

File	Role	Link
src/datanode/src/region_server.rs	Core RegionServer implementation (read & write orchestration).	region_server.rs
src/servers/src/grpc/region_server.rs	gRPC RegionServerHandler that receives RegionRequest messages.	grpc/region_server.rs
src/servers/src/grpc/builder.rs	Builder wiring that registers the RegionServer service with the gRPC server.	builder.rs
src/table/src/catalog.rs (via NameAwareCatalogList)	Provides the region‑specific catalog for query planning.	catalog.rs
src/query/src/engine.rs (QueryEngine)	Executes logical plans produced by the RegionServer’s read path.	query engine
src/metric_engine/src/engine.rs (MetricEngine)	Implements put_regions_batch for fast batch inserts.	metric_engine.rs

Code Examples

Performing a Remote Read (Client Side)

Clients send Arrow Flight Ticket messages containing encoded QueryRequest protobufs. The following Rust example demonstrates building and sending a read request:

use arrow_flight::Ticket;
use greptime_proto::v1::region::QueryRequest;
use greptime_proto::v1::region::query_request::Header;
use greptime_proto::v1::region::query_request::TracingContext;

// Build a QueryRequest (plan already encoded as protobuf bytes)
let request = QueryRequest {
    header: Some(Header {
        tracing_context: Some(TracingContext { /* … */ }),
        ..Default::default()
    }),
    region_id: my_region_id,
    plan: encoded_logical_plan, // `Bytes` from the planner
};

// Encode into a Flight Ticket
let ticket = Ticket {
    ticket: request.encode_to_vec().into(),
};

// Call Arrow Flight `do_get`
let mut client = FlightServiceClient::connect("grpc://127.0.0.1:4001").await?;
let response = client.do_get(Request::new(ticket)).await?;
let stream = response.into_inner(); // `FlightRecordBatchStream`
while let Some(batch) = stream.message().await? {
    // process RecordBatch …
}

The request routes to RegionServer::do_get, which delegates to handle_remote_read → handle_read → QueryEngine for execution.

Inserting Rows (Client Side)

Write requests use the GreptimeDB gRPC protocol. The following example constructs a batch insert request:

use greptime_proto::v1::region::{RegionRequest, PutRequest};
use greptime_proto::v1::region::put_request::RowGroup;

// Build a PutRequest with rows encoded as protobuf
let put = PutRequest {
    rows: vec![RowGroup { /* … */ }],
    // ... other fields like version, ttl, etc.
};

let request = RegionRequest::Put(put);
let body = region_request::Body::Inserts(vec![(region_id, request)]);

// Send via gRPC RegionServer
let mut client = GreptimeRegionServiceClient::connect("http://127.0.0.1:4001").await?;
let response = client.handle(Request::new(body)).await?;
println!("Affected rows: {}", response.get_ref().affected_rows);

This gRPC call triggers RegionServerHandler::handle, which dispatches to handle_requests_in_parallel. When the target engine is MetricEngine, the server optimizes the batch via put_regions_batch.

Summary

• Dual-protocol design: The RegionServer exposes Arrow Flight for high-performance analytical reads and GreptimeDB gRPC for transactional writes, unifying access through a single datanode endpoint.
• Read path: do_get decodes Flight Tickets into QueryRequest, acquires parallelism permits, injects region-aware catalogs via NameAwareCatalogList, and executes through the QueryEngine, returning compressed Arrow RecordBatches.
• Write path: The gRPC handler routes requests by type—parallel execution for inserts/deletes via handle_requests_in_parallel, serial execution for DDL via handle_requests_in_serial, with special batch optimization for MetricEngine through put_regions_batch.
• Safety mechanisms: Region state management (Registering, Deregistering, Ready) prevents concurrent modifications, while RegionServerParallelism semaphore limits protect against overload.

Frequently Asked Questions

What protocol does the RegionServer use for read requests?

The RegionServer exposes read capabilities via the Arrow Flight protocol. Clients send Ticket messages containing encoded QueryRequest protobufs to the do_get endpoint. This design leverages Arrow's efficient columnar format for zero-copy data transfer and high-throughput analytical queries.

How does the RegionServer handle high-concurrency write workloads?

For write-heavy workloads, the RegionServer uses handle_requests_in_parallel to process multiple region requests concurrently using try_join_all. Each request acquires a permit from the RegionServerParallelism semaphore to prevent resource exhaustion. Additionally, when all requests are Put operations targeting the MetricEngine, the server invokes put_regions_batch to optimize throughput by processing the entire batch in a single call rather than iterating per-region.

What is the MetricEngine batch optimization?

The MetricEngine batch optimization is a specialized fast path for high-cardinality metric ingestion. When the RegionServer detects that all requests in a batch are Put operations and the target engine is MetricEngine, it bypasses the standard per-region loop and calls MetricEngine::put_regions_batch. This reduces overhead and improves ingestion performance for monitoring workloads, returning a single RegionResponse with the total affected row count.

How does the RegionServer protect against overload?

The RegionServer implements a concurrency limiter via the RegionServerParallelism configuration. Before executing read or write requests, the server attempts to acquire a permit from a Tokio semaphore with a configurable timeout. If the permit cannot be acquired or if the server is in a suspended state (is_suspended()), the request returns an error immediately, protecting the node from memory exhaustion and CPU saturation during traffic spikes.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}