How Data Distribution Works in GreptimeDB Distributed Mode with Partition Expressions

GreptimeDB distributes data across multiple regions (storage shards) by evaluating partition expressions that define value ranges for each region, ensuring efficient write routing and read pruning through three coordinated components: PartitionRuleManager, RowSplitter, and ConstraintPruner.

GreptimeDB is an open-source time-series database built for cloud-native scalability. When operating in distributed mode, it relies on data distribution in distributed mode with partition expressions to shard data across multiple regions, ensuring both high availability and query performance. This article analyzes the source code in greptimeteam/greptimedb to explain the exact mechanism of how partition expressions drive data distribution.

The Three Core Components of Data Distribution

GreptimeDB's data distribution relies on three tightly-coupled components that work together to route writes and prune reads based on partition expressions.

PartitionRuleManager

The PartitionRuleManager (in src/partition/src/manager.rs) discovers partition rules for tables and caches the mapping between region numbers and their JSON-encoded partition expressions. It provides the find_table_partition_rule method to retrieve cached PhysicalPartitionInfo and builds PartitionInfo objects containing the expressions for each region.

RowSplitter

The RowSplitter (in src/partition/src/splitter.rs) handles write-side sharding. It evaluates each incoming row's partition columns against the cached rule via the split method, returning a HashMap<RegionNumber, Rows> that groups rows by their target region.

ConstraintPruner and PredicateExtractor

For read operations, ConstraintPruner (in src/query/src/dist_plan/region_pruner.rs) and PredicateExtractor (in src/query/src/dist_plan/predicate_extractor.rs) work together to eliminate irrelevant regions. The extractor converts DataFusion filter predicates into PartitionExpr objects, while the pruner checks for atomic overlap between query predicates and each region's stored partition expression.

Discovering and Caching Partition Rules

When a table is created with a PARTITION BY clause, GreptimeDB stores the partition rule in the meta service. The PartitionRuleManager retrieves this metadata and builds a mapping of region identifiers to their corresponding partition expressions.

In src/partition/src/manager.rs, the find_table_partition_rule method fetches the cached information:

// src/partition/src/manager.rs
pub async fn find_table_partition_rule(
    &self,
    table_info: &TableInfo,
) -> Result<(PartitionRuleRef, HashMap<RegionNumber, Option<u64>>)> {
    // … fetch the rule from the cached `PhysicalPartitionInfo`
}

The manager parses the JSON partition expressions when building the region mapping:

// src/partition/src/manager.rs: create_partitions_from_region_routes
let partition_expr = PartitionExpr::from_json_str(&r.region.partition_expr())?;

Write Path: Routing Rows with RowSplitter

During write operations, GreptimeDB evaluates each row's partition columns to determine the appropriate target region.

The write path begins with PartitionRuleManager::split_rows, which delegates to RowSplitter:

// src/partition/src/manager.rs: split_rows
let result = RowSplitter::new(partition_rule)
    .split(rows)?               // 👉 evaluate each row ↔ partition expression

In src/partition/src/splitter.rs, the RowSplitter::split method evaluates the partition columns and groups rows:

// src/partition/src/splitter.rs (simplified)
pub fn split(&self, rows: Rows) -> Result<HashMap<RegionNumber, Rows>> {
    // For each row:
    //   evaluate `partition_rule` → region_number
    //   push the row into the corresponding bucket
}

Consider inserting rows into a table metrics partitioned by a ts column:

use greptime_client::Client;               // hypothetical client crate
use greptime_client::Row;                  // row builder

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let client = Client::with_urls(vec!["http://localhost:4000"]).await?;
    // Table `metrics` is partitioned by `ts` column (time range)
    let rows = vec![
        Row::new().add("ts", 1630454400_i64).add("value", 42.0_f64),
        Row::new().add("ts", 1630540800_i64).add("value", 36.5_f64),
    ];
    client.insert("metrics", rows).await?; // ← PartitionRuleManager splits automatically
    Ok(())
}

Behind the scenes: client.insert calls PartitionRuleManager::split_rows, which uses RowSplitter to route the first row to region 1 (ts < 1630500000) and the second row to region 2 (ts >= 1630500000).

Read Path: Pruning Regions with ConstraintPruner

On the read path, GreptimeDB minimizes data transfer by pruning irrelevant regions using the ConstraintPruner and PredicateExtractor. This ensures queries only scan regions whose partition expressions overlap with the query predicates.

The DistExtensionPlanner in src/query/src/dist_plan/planner.rs orchestrates this process:

// src/query/src/dist_plan/planner.rs → get_regions
let partition_expressions = PredicateExtractor::extract_partition_expressions(
    logical_plan,
    &partition_columns,
)?;

In src/query/src/dist_plan/predicate_extractor.rs, the PredicateExtractor walks the logical plan and converts DataFusion expressions into PartitionExpr objects, keeping only predicates that involve partition columns (AND-only fragments are kept; OR-mixed fragments are dropped).

The extracted expressions are passed to ConstraintPruner::prune_regions in src/query/src/dist_plan/region_pruner.rs:

// src/query/src/dist_plan/region_pruner.rs
let candidate_regions = ConstraintPruner::prune_regions(
    &partition_expressions,
    &partitions,
    partition_column_types,
)?;

For example, given this SQL query:

-- only rows where ts is between 2021‑08‑31 and 2021‑09‑02
SELECT ts, value FROM metrics
WHERE ts >= 1630454400 AND ts < 1630627200;

The ConstraintPruner identifies that only regions whose stored expressions overlap with the interval [1630454400, 1630627200) need to be scanned, skipping all unrelated regions.

End-to-End Data Flow Example

The complete flow from table creation to query execution works as follows:

1. Table Creation: A PARTITION BY clause creates a PartitionRule persisted in the meta service (e.g., ts >= 0 AND ts < 1000 for region 1).

2. Insert: When inserting rows with timestamps 1630454400 and 1630540800, the RowSplitter routes the first row to region 1 and the second to region 2 based on the partition expressions.

3. Query: For a query with WHERE ts >= 1630454400 AND ts < 1630627200, the ConstraintPruner identifies that only region 1 and region 2 overlap with this range, skipping unrelated regions.

The same partition expression (JSON string) stored in region metadata (api::v1::meta::Partition) serves as the single source of truth for both write routing and read pruning.

Key Source Files

The following files implement the data distribution logic in greptimeteam/greptimedb:

• src/partition/src/manager.rs: Central manager for partition rules, region-to-expression mapping, and row splitting entry point.
• src/partition/src/expr.rs: Definition of PartitionExpr, JSON conversion, and operand handling.
• src/partition/src/splitter.rs: Implements RowSplitter::split for write-side sharding logic.
• src/query/src/dist_plan/predicate_extractor.rs: Extracts and sanitizes partition predicates from logical plans.
• src/query/src/dist_plan/region_pruner.rs: Core pruning algorithm matching query predicates with region expressions.
• src/query/src/dist_plan/planner.rs: Orchestrates region discovery by invoking the extractor and pruner.

Summary

• Partition expressions stored as JSON in region metadata serve as the single source of truth for data distribution in GreptimeDB's distributed mode.
• PartitionRuleManager caches the mapping between regions and their partition expressions, enabling efficient lookup during both writes and reads.
• RowSplitter evaluates partition columns during write operations to route rows to the correct regions based on the cached rules.
• ConstraintPruner and PredicateExtractor work together on the read path to eliminate irrelevant regions, ensuring queries only scan data that potentially matches the query predicates.
• The same partition expression format is used consistently for both write routing and read pruning, maintaining consistency across the distributed cluster.

Frequently Asked Questions

What is a partition expression in GreptimeDB?

A partition expression is a JSON-encoded logical expression that defines the value range belonging to a specific region. For example, an expression like ts >= 0 AND ts < 1000 indicates that a region stores only rows where the ts column falls within that range. These expressions are stored in the meta service and evaluated by PartitionRuleManager to route writes and by ConstraintPruner to filter reads.

How does GreptimeDB decide which region receives a write?

When rows are inserted, the RowSplitter evaluates each row's partition columns against the cached partition expressions via the split method in src/partition/src/splitter.rs. It groups rows by matching region and returns a HashMap<RegionNumber, Rows>, which the write path uses to forward each batch to the appropriate region leader.

Can queries skip regions that don't contain relevant data?

Yes. During query planning, the ConstraintPruner (implemented in src/query/src/dist_plan/region_pruner.rs) compares the query's filter predicates against each region's partition expression using atomic overlap detection. It eliminates regions whose defined ranges cannot intersect with the query predicates, ensuring only relevant regions are scanned.

What happens if a partition expression involves multiple columns?

GreptimeDB supports composite partition keys. The PredicateExtractor in src/query/src/dist_plan/predicate_extractor.rs identifies predicates involving any partition columns, and the ConstraintPruner evaluates the combined constraints. AND-only fragments are fully utilized for pruning, while OR-mixed fragments involving partition columns are conservatively handled to ensure correctness without missing potential matches.

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