Supports multi-region

[v0y4g3r]

What type of enhancement is this?

Tech debt reduction

What does the enhancement do?

Currently in GreptimeDB, only one region can be created for a table on one datanode, which will cause data corruption when the number of table partitons exceeds the number of datanodes.

In fact, table engine supports creating multiple regions for one table, but we have to go through all those components on datanodes whether they can accommodate multiple regions.

Implementation challenges

To support multiple regions on datanode, both standalone and distributed mode should be considered.

• In distributed mode, requests are parsed on frontend, and frontend forward queries to datanode. Thus, datanode's incoming requests must carry region info, for example, the region to insert, regions to query from etc. It should be relatively easy to add this parameters.
• While in standalone mode, things get a bit tricky. If datanode has only one region, then query engine is not aware of partition rules. But if datanode has multiple regions for one table, then query engine should associate requests with corresponding regions according to partition rules. There are two problems here:
• In standalone mode, catalog manager does not support storing partition rules, we must add an options field to SystemCatalogTable
  

  greptimedb/src/catalog/src/system.rs

  Line 49 in 8402ef9

  pub struct SystemCatalogTable {

• Find regions for queries according to partition rules. Currently it's implemented in DistTable. But in standalone mode, requests are directly forwarded to datanode. Thus we need to extract the partition logic to some separate struct to reused it in both distributed mode and standalone mode.
  

  greptimedb/src/frontend/src/table.rs

  Lines 85 to 95 in 8402ef9

  	async fn insert(&self, request: InsertRequest) -> table::Result<usize> {
  	let partition_rule = self.find_partition_rule().await.map_err(TableError::new)?;
  	let spliter = WriteSpliter::with_partition_rule(partition_rule);
  	let inserts = spliter.split(request).map_err(TableError::new)?;
  	let result = match self.dist_insert(inserts).await.map_err(TableError::new)? {
  	client::ObjectResult::Select(_) => unreachable!(),
  	client::ObjectResult::Mutate(result) => result,
  	};
  	Ok(result.success as usize)
  	}

Future work

• Scan on schema change: when schema change happens while scanning multiple regions on one datanode, currently it just raises an error, we may need to try to adapt to schema change.
  

  greptimedb/src/mito/src/table.rs

  Lines 178 to 186 in 5d62e19

  	ensure!(
  	first_schema.version() == schema.version(),
  	RegionSchemaMismatchSnafu {
  	table: common_catalog::format_full_table_name(
  	&table_info.catalog_name,
  	&table_info.schema_name,
  	&table_info.name
  	)
  	}

• We need to find a way to pass region filters to datanode through substrait plans so that we don't have to iterate all regions on some datanode Pass region filters through substrait plans #926
• DROP TABLE SQL is not currently implemented in parser, we need to support drop all regions on datanode. Support drop table clause #497
• We need to adapt DDL operations to multi regions to procedure framework to allow fault recovery as soon as Tracking issue for the procedure framework #286 finishes.

[v0y4g3r]

MitoTable should be modified to accommodate multiple regions, and it's create/open/insert/select methods must explicitly provide regions.

Currently all partition rules are parsed in frontend and datanode does not validate if data inserted satisfies partition rule, but we may add this validation once repartition is supported.

[v0y4g3r]

Maybe we should add an regions: Option<Vec<RegionNumber>> argument.

Vec<RegionNumber> should be adequate as None and an empty vector should represent the same thing.

Also, if there're regions with different schema in a table, then what will be the schema for record batches yielded? Can we compare the versions of schemas of all regions and find the schema with the min version?

[evenyag]

Can we compare the versions of schemas of all regions and find the schema with the min version

This might be the most applicable way before #275 is done. BTW, this reminds me that we need to move the version field somewhere else, as mentioned in #349

[MichaelScofield]

Looks like region failover(#1126) also requires multi region support.

[fengjiachun]

Close this issue as version 0.4 already supports multi-region.