How GreptimeDB's SQL Parser Handles Partition Routing for Query Execution

GreptimeDB routes SQL queries to specific storage partitions by detecting ordering patterns during the parsing and optimization phases, converting them into distribution hints that the physical execution engine uses to hash on the hidden __tsid column and target only relevant regions.

In the greptimeteam/greptimedb repository, the SQL parser partition routing mechanism transforms logical query plans into efficient physical execution strategies. This process identifies time-series distribution patterns during the initial parsing stage and propagates routing information through the optimizer to ensure scans only touch the regions containing relevant data.

Stage 1: Detecting Distribution Patterns During SQL Parsing

The partition routing logic begins in the query optimizer's hint collection phase. As the SQL parser builds the LogicalPlan, the ScanHintVisitor traverses the plan tree looking for specific ordering patterns that imply a per-series data distribution.

When the visitor encounters a Sort node with exactly two columns—first the hidden __tsid (time-series identifier) and second the time index—it records a distribution hint. In src/query/src/optimizer/scan_hint.rs (lines 52-62), the visitor checks:

// Inside ScanHintVisitor::f_down()
if sort_cols.len() == 2
    && sort_cols[0].name == DATA_SCHEMA_TSID_COLUMN_NAME
    && sort_cols[1].name == time_index_name
{
    // Tell the table provider that the query can be executed per-series
    adapter.with_distribution(TimeSeriesDistribution::PerSeries);
}

This detection allows the engine to recognize queries that can benefit from partitioned execution based on series ID rather than requiring a full table scan.

Stage 2: Embedding Hints into the Scan Request

Once the ScanHintVisitor identifies a per-series pattern, it stores the TimeSeriesDistribution::PerSeries hint inside a dummy table provider. This provider later constructs the ScanRequest struct defined in src/store-api/src/storage/requests.rs, which carries the optional distribution field:

pub struct ScanRequest {
    // ... other fields ...
    /// Optional hint for the distribution of time-series data.
    pub distribution: Option<TimeSeriesDistribution>,
}

The distribution field acts as a bridge between the logical plan analysis and the physical execution strategy, instructing downstream components about the optimal data partitioning approach.

Stage 3: Physical Plan Construction and Partitioning

During physical plan generation, the RegionScanExec node reads the distribution hint to determine its partitioning strategy. In src/table/src/table/scan.rs (lines 94-108), the RegionScanExec::new method matches on request.distribution:

let partitioning = match request.distribution {
    Some(TimeSeriesDistribution::PerSeries) => {
        // Use the hidden __tsid column for hash-partitioned reads
        Partitioning::Hash(vec![Arc::new(tsid_col) as _], num_output_partition)
    }
    _ => Partitioning::UnknownPartitioning(num_output_partition),
};

When the hint is present, the execution node configures hash partitioning on the __tsid column. This ensures that data with the same series identifier routes to the same physical partition, enabling targeted region scans.

Stage 4: Optimizer Propagation Through Merge Scans

The PassDistribution optimizer rule ensures that distribution requirements flow through the entire physical plan tree. Located in src/query/src/optimizer/pass_distribution.rs (lines 66-88), this pass specifically handles MergeScanExec nodes:

if let Some(merge_scan) = plan.as_any().downcast_ref::<MergeScanExec>()
    && let Some(distribution) = current_req.as_ref()
    && let Some(new_plan) = merge_scan.try_with_new_distribution(distribution.clone())
{
    return Ok(Arc::new(new_plan));
}

By propagating the distribution hint to merge scan operations, the optimizer guarantees that all downstream physical operators respect the same routing logic, maintaining partition locality throughout the query execution pipeline.

Stage 5: Runtime Region Routing

At execution time, the RegionScanner implementation in src/mito2/src/read/scan_region.rs consumes the distribution hint to perform the actual routing decision. The scanner matches on the request's distribution field:

match request.distribution {
    Some(TimeSeriesDistribution::PerSeries) => {
        // Hash the series id to a region id
        let region_id = hash_series_id(&tsid) % peer_count;
        // Send the scan request only to that region
        send_to_region(region_id, request);
    }
    _ => {
        // Full scan across all regions
        broadcast_scan(request);
    }
}

This runtime routing ensures that when PerSeries distribution is specified, the engine computes the target region by hashing the __tsid value and contacts only the region(s) owning that series, dramatically reducing I/O overhead.

Practical Example: Triggering Per-Series Routing

To leverage partition routing in your queries, structure your SQL to order by the series identifier followed by the timestamp:

-- This pattern triggers PerSeries distribution hint
SELECT host, value
FROM cpu_metrics
ORDER BY __tsid, ts;

This ordering pattern signals the ScanHintVisitor to apply the PerSeries distribution, allowing the execution engine to route the scan to specific partitions rather than broadcasting across all regions.

Summary

• The SQL parser detects partition routing opportunities by analyzing ORDER BY clauses for __tsid, time patterns in src/query/src/optimizer/scan_hint.rs.
• Detected patterns become TimeSeriesDistribution::PerSeries hints attached to ScanRequest structs in src/store-api/src/storage/requests.rs.
• The physical planner (RegionScanExec::new in src/table/src/table/scan.rs) translates hints into hash partitioning strategies on the __tsid column.
• The optimizer propagates distribution requirements through MergeScanExec nodes via PassDistribution in src/query/src/optimizer/pass_distribution.rs.
• Runtime routing in src/mito2/src/read/scan_region.rs uses these hints to target specific regions, avoiding unnecessary full-table scans.

Frequently Asked Questions

How does the SQL parser determine which partition should handle a query?

The SQL parser does not directly assign partitions. Instead, the ScanHintVisitor analyzes the logical plan for ordering patterns—specifically ORDER BY __tsid, time—and attaches a TimeSeriesDistribution::PerSeries hint to the scan request. The physical execution layer later uses this hint to hash the __tsid value and route to the appropriate region.

What is the role of the __tsid column in partition routing?

The __tsid column serves as the hidden series identifier that enables hash-based partition routing. When the physical plan detects TimeSeriesDistribution::PerSeries, it configures Partitioning::Hash using __tsid as the hash key, ensuring all data for a specific time series routes to the same partition and enabling targeted region scans.

How does the optimizer ensure distribution hints propagate through complex queries?

The PassDistribution optimizer rule traverses the physical plan tree and injects distribution requirements into MergeScanExec nodes. This ensures that when a query contains multiple scan operations or aggregation phases, all downstream nodes inherit the same PerSeries distribution logic, maintaining consistent partition routing throughout the execution pipeline.

What happens when a query does not provide a distribution hint?

If the ScanHintVisitor does not detect the specific ordering pattern, the ScanRequest carries distribution: None. In this case, RegionScanExec defaults to Partitioning::UnknownPartitioning, and the runtime scanner falls back to a full scan across all regions rather than targeting specific partitions.

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